ストリーミング取り込みとレプリケーションターゲット

• •Amazon Kinesis Data Firehose
• •Amazon Kinesis Streams
• •Amazon S3
• •Databricks
• •フラットファイル
• •Google BigQuery V2
• •Google Cloud Storage
• •Google PubSub
• •JDBC V2
• • Kafka
• - Apache Kafka
• - Confluent Kafka
• - Amazon Managed Streaming（Amazon MSK）
• •Microsoft Azure Data Lake Store Gen2
• •Microsoft Azure Event Hubs

Amazon Kinesis Data Firehoseターゲット

• •IAMユーザーがAWS Kinesis DataFirehoseサービスを使用するために必要なIAM権限を持つAWSアカウントを作成します。
• •Firehose配信ストリームを定義します。ソースをDirect PUTまたはその他のソースとして設定します。
• •ユーザーが書き込んでいるターゲットに基づいて、IAMユーザーの資格情報に必要な権限を付与します。
権限のリストについては、https://docs.aws.amazon.com/firehose/latest/dev/controlling-access.html#access-to-firehoseのAWSドキュメントを参照してください。

Amazon Kinesis Streamsターゲット

1. 1Amazon Kinesisコンソールで、Amazon Kinesisストリームを作成して設定します。
2. 2Amazon Web Services（AWS）Identity and Access Management（IAM）サービスで、ユーザーを作成します。
3. 3ユーザー作成プロセス中に生成されたアクセスキーとシークレットアクセスキーをダウンロードします。
4. 4Kinesisストリームへの書き込み権限を持つグループにユーザーを関連付けます。

Amazon S3ターゲット

データのパーティション化

ストリーミング取り込みとレプリケーションタスクは、Amazon S3 V2ターゲットにパーティションを作成し、データをパーティションに書き込むことができます。パーティション化を使用するには、タスクがパーティションを作成する際に従うパーティション化間隔を選択する必要があります。選択した時間間隔に基づいて、ストリーミング取り込みとレプリケーションジョブは受信メッセージをAmazon S3 V2バケットの/<オブジェクト名>/<年>/<月>/<日>/<時間>/<分>パーティションに保存します。ストリーミング取り込みとレプリケーションジョブは、Amazon S3バケットにタイムスタンプ階層フォルダを追加します。

タスクを設定するときに、次の方法でパーティション化を有効にすることができます。

• - ［オブジェクト名/式］フィールドでオブジェクト名に${Timestamp}式を追加し、時間間隔を選択します。
例えば、［オブジェクト名/式］フィールドで、/streaming/${Timestamp}と入力し、5分のパーティション化間隔を選択してから、2022年6月8日の15:20にストリーミング取り込みとレプリケーションタスクを実行したとします。ストリーミング取り込みとレプリケーションジョブは、受信メッセージをAmazon S3 V2バケットの/streaming/2022/06/08/15/20パーティションに保存します。このジョブは、次の時間間隔の際にストリーミングされるデータを、時間フォルダ（2022/06/08/15/25）に保存します。
注: ［オブジェクト名/式］フィールドで、オブジェクト名に${Timestamp}式を追加し、パーティション化間隔を選択しなかった場合、ストリーミング取り込みとレプリケーションジョブはオブジェクトをAmazon S3バケットの0/フォルダに保存します。
• - ［オブジェクト名/式］フィールドで正規表現を使用し、パーティション化間隔を選択します。正規表現に${Timestamp}表現を追加する必要はありません。
例えば、2022年6月18日の12:10にストリーミング取り込みとレプリケーションタスクを、正規表現$"SourceTable":"((.*?))"$と、5分のパーティション化間隔で実行したとします。受信データは{"SourceTable":"xyz"}です。ストリーミング取り込みとレプリケーションジョブは、xyzオブジェクトをAmazon S3バケットの2022/06/18/12/10フォルダ階層に保存します。このジョブは、次の時間間隔の際にストリーミングされるデータを、時間フォルダ（2022/06/08/15/25）に保存します。

Databricksターゲット

• •Azure Data Lake Storage（ADLS）Gen2
• •Amazon Web Services（AWS）S3

• •すべてのストリーミングソースからDatabricksテーブルにバルクデータを取り込む
• •すべてのストリーミングソースからの変更データキャプチャ（CDC）をマージし、Databricksテーブルに書き込む

フラットファイルターゲット

Google BigQuery V2ターゲット

• •Google BigQueryにアクセスするためのGoogleサービスアカウントを持っていることを確認してください。
• •サービスアカウントにclient_email、project_id、private_key、およびregionIDの値があることを確認してください。Google BigQuery V2接続を作成するときに、対応する［サービスアカウントID］、［プロジェクトID］、［サービスアカウントキー］、および［リージョンID］接続プロパティに値を入力します。
• •private_key_idプロパティとclient_idプロパティを［オプションのプロパティを指定］フィールドで指定する必要があります。これらのパラメータを指定すると、Google BigQuery V2接続テストは失敗します。ただし、失敗したテスト接続を無視して、ストリーミング取り込みとレプリケーションジョブを実行できます。次の形式を使用します。
"private_key_id":"<private key ID>" and "client_id":"<client ID>"
• • Google BigQuery接続のタイムアウト間隔を設定する場合は、接続プロパティの［オプションのプロパティを指定］フィールドでタイムアウト間隔プロパティを指定します。次の形式を使用します。
"timeout": "<timeout_interval_in_seconds>"
• •ストリーミング取り込みとレプリケーションタスクをデプロイする前に、データを書き込むためのテーブルが存在している必要があります。
• •ターゲットテーブルを含むGoogle BigQueryデータセットへの読み取りおよび書き込みアクセス権が必要です。

Google Cloud Storage V2ターゲット

• •プロジェクト。Google Cloud Storageでは、すべてのリソースがプロジェクト内に格納されます。プロジェクトは、請求の詳細とユーザーの詳細を格納する最上位のコンテナです。プロジェクトは複数作成できます。プロジェクトには、一意のプロジェクト名、プロジェクトID、およびプロジェクト番号があります。
• •バケット。各バケットは、データを格納するコンテナのように機能します。バケットを使用して、データを整理し、データにアクセスできます。複数のバケットを作成できますが、バケットをネストすることはできません。バケット内に複数のフォルダを作成でき、フォルダをネストすることもできます。アクセス制御リストを定義して、オブジェクトとバケットを管理できます。アクセス制御リストは、権限と範囲のエントリで構成されています。権限は、読み取りまたは書き込み操作を実行するためのアクセスを定義します。範囲は、操作を実行できるユーザーまたはグループを定義します。
• •オブジェクト。オブジェクトは、Google Cloud Storageにアップロードするデータを構成します。バケット内にオブジェクトを作成できます。オブジェクトは、オブジェクトデータとオブジェクトメタデータコンポーネントで構成されます。オブジェクトデータは、Google Cloud Storageに保存するファイルです。オブジェクトメタデータは、オブジェクトの特性を説明する名前と値のペアのコレクションです。

1. 1Google Cloud StorageにアクセスするためのGoogleサービスアカウントを持っていることを確認してください。
2. 2サービスアカウントのclient_email、project_id、private_keyの値があることを確認してください。AdministratorでGoogle Cloud Storage接続を作成するときに、これらの詳細を入力する必要があります。
3. 3サービスアカウントでGoogle Cloud Storage JSON APIが有効になっていることを確認してください。Google Cloud Storage V2コネクタは、Google APIを使用してGoogle Cloud Storageと統合します。
4. 4ターゲットファイルを含むGoogle Cloud Storageバケットへの書き込みアクセスがあることを確認してください。
5. 5組織でCloudera CDHまたはHortonworks HDPパッケージを使用するためのライセンスが有効になっていることを確認してください。

Google PubSubターゲット

• •組織にGoogle PubSubコネクタライセンスがある。
• •Google PubSubにアクセスするためのGoogleサービスアカウントのJSONキーがある。
• •Googleサービスアカウントのclient_email、project_id、private_keyの値がある。AdministratorでGoogle PubSub接続を作成するときに、これらの詳細が必要です。

JDBC V2ターゲット

• •ストリーミング取り込みタスクをデプロイする前に、データを書き込むためのテーブルが存在している必要があります。
• • データベースドライバファイルを次のディレクトリにコピーします。
<Secure Agentインストールディレクトリ>/apps/Streaming_Ingestion_Agent/ext

Kafkaターゲット

Microsoft Azure Data Lake Storage Gen2ターゲット

Microsoft Azure Event Hubsターゲット