DataWorks:Data Integration でのリアルタイム同期タスクの設定

1. ソースの設定

ページの上部で、Kafka データソースをクリックします。次に、[Kafka ソース情報] を編集します。

1. [Kafka ソース情報] セクションで、Kafka データソースから同期するトピックを選択します。

   他の設定にはデフォルト値を使用するか、必要に応じて変更できます。パラメーターの詳細については、Kafka の公式ドキュメントをご参照ください。

2. 右上隅にある [データサンプリング] をクリックします。

   表示されるダイアログボックスで、[開始時刻] と [サンプル数] を設定し、[サンプリング開始] をクリックします。この操作により、指定された Kafka トピックからデータがサンプリングされます。トピック内のデータをプレビューできます。このプレビューは、後続のデータ処理ノードのデータプレビューと可視化設定の入力として提供されます。

3. [出力フィールドの設定] セクションで、必要に応じて同期するフィールドを選択します。

   デフォルトでは、Kafka は 6 つのフィールドを提供します。

   フィールド名	説明
   __key__	Kafka レコードのキー。
   __value__	Kafka レコードの値。
   __partition__	Kafka レコードが配置されているパーティション番号。パーティション番号は 0 から始まる整数です。
   __headers__	Kafka レコードのヘッダー。
   __offset__	パーティション内での Kafka レコードのオフセット。オフセットは 0 から始まる整数です。
   __timestamp__	Kafka レコードの 13 桁の UNIX タイムスタンプ (ミリ秒単位)。

   後続のデータ処理ノードで、さらに多くのフィールド変換を実行することもできます。

2. データ処理ノードの編集

アイコンをクリックして、データ処理メソッドを追加します。利用可能なメソッドは、データマスキング、文字列置換、データフィルタリング、JSON 解析、フィールドの編集と割り当て の 5 つです。これらのメソッドを目的の順序で配置できます。実行時、データ処理メソッドは指定された順序で実行されます。

データ処理ノードを設定した後、右上隅の [データ出力のプレビュー] をクリックできます：

1. 入力データの下のテーブルで、前の [データサンプリング] ステップの結果を表示できます。[アップストリーム出力の再取得] をクリックして結果を更新します。

2. アップストリームノードからの出力がない場合は、[手動でデータを構築] をクリックして前の出力をシミュレートすることもできます。

3. [プレビュー] をクリックして、アップストリームステップからの出力データがデータ処理コンポーネントによって処理された後のデータを表示します。

3. ターゲットの設定

ページの上部で、MaxCompute データターゲットをクリックします。次に、MaxCompute ターゲット情報を編集します。

1. [MaxCompute ターゲット情報] セクションで、トンネルリソースグループを選択します。デフォルトは「パブリック転送リソース」で、これは MaxCompute が提供する無料クォータです。

2. ターゲットテーブルに [テーブルの自動作成] を行うか、[既存のテーブルを使用] するかを指定します。

   1. テーブルの自動作成を選択した場合、デフォルトでソーステーブルと同じ名前のテーブルが作成されます。ターゲットテーブル名は手動で変更できます。

   2. 既存のテーブルを使用することを選択した場合は、ドロップダウンリストからターゲットテーブルを選択します。

3. (オプション) テーブルスキーマを編集します。

   [テーブルの自動作成] を選択した場合、[テーブルスキーマの編集] をクリックします。表示されるダイアログボックスで、ターゲットテーブルのスキーマを編集します。また、[アップストリームの出力列に基づいてテーブルスキーマを再生成] をクリックして、アップストリームノードの出力列に基づいてテーブルスキーマを自動的に生成することもできます。自動生成されたスキーマで列を選択し、それをプライマリキーとして設定できます。

4. フィールドマッピングを設定します。

   1. システムは、[同じ名前のマッピング] の原則に基づいて、アップストリームの列をターゲットテーブルの列に自動的にマッピングします。必要に応じてマッピングを調整できます。1 つのアップストリーム列を複数のターゲット列にマッピングできますが、複数のアップストリーム列を 1 つのターゲット列にマッピングすることはできません。アップストリーム列がターゲット列にマッピングされていない場合、そのデータはターゲットテーブルに書き込まれません。

   2. Kafka フィールドについては、カスタムの JSON 解析を設定できます。データ処理コンポーネントを使用して value フィールドのコンテンツを取得します。これにより、より詳細なフィールド設定が可能になります。

      

5. (オプション) パーティションを設定します。

   1. [時間ベースの自動パーティション分割] は、ビジネス時間 (この場合は _timestamp フィールド) に基づいてパーティションを作成します。第 1 レベルのパーティションは年、第 2 レベルのパーティションは月、というようになります。

   2. [フィールドコンテンツによる動的パーティション分割] は、ソーステーブルのフィールドをターゲットの MaxCompute テーブルのパーティションフィールドにマッピングします。これにより、ソースフィールドに特定のデータを含む行が、MaxCompute テーブルの対応するパーティションに書き込まれるようになります。

アラート設定

タスクエラーによるデータ同期遅延を防ぐために、単一テーブルのリアルタイム同期タスクにアラートポリシーを設定できます。

1. ページの右上隅にある [アラート設定] をクリックして、タスクのアラート設定ページを開きます。

2. [アラートの追加] をクリックして、アラートルールを設定します。データ遅延、フェールオーバーイベント、タスクステータス、データ定義言語 (DDL) の変更、タスクリソース使用率などのメトリックを監視するアラートトリガーを設定できます。指定されたしきい値に基づいて、CRITICAL または WARNING のアラートレベルを設定できます。

3. アラートルールを管理します。

   既存のアラートルールについては、アラートスイッチを使用して有効または無効にできます。また、アラートレベルに基づいて異なる担当者にアラートを送信することもできます。

高度なパラメーター設定

同期タスクには、詳細な設定を行うための高度なパラメーターが用意されています。システムはデフォルト値を提供しており、ほとんどの場合、変更する必要はありません。変更する場合：

1. ページの右上隅にある [高度なパラメーター] をクリックして、高度なパラメーター設定ページを開きます。

2. [ランタイム設定の自動構成] を false に設定します。

3. ツールチップに基づいてパラメーター値を変更します。各パラメーターの説明は、その名前の横に表示されます。

リソースグループの設定

ページの右上隅にある [リソースグループの設定] をクリックして、現在のタスクで使用されているリソースグループを表示および切り替えることができます。