すべてのプロダクト
Search

このプロダクト

    DataWorks:Kafka から OSS への単一テーブルのリアルタイム同期

    ドキュメントセンター

    DataWorks:Kafka から OSS への単一テーブルのリアルタイム同期

    最終更新日:Apr 10, 2026

    このトピックでは、Data Integration を使用して Kafka ソースから OSS データレイクにデータをリアルタイムで同期する方法について説明します。

    制限事項

    Kafka のバージョンは 0.10.2 から 2.2.0 (両端を含む) の範囲内である必要があります。

    前提条件

    操作手順

    1. 同期タスクタイプの選択

    1. Data Integration ページに移動します。

      DataWorks コンソールにログインします。上部のナビゲーションバーで、目的のリージョンを選択します。左側のナビゲーションウィンドウで、データ統合 > データ統合を選択します。表示されたページで、ドロップダウンリストから目的のワークスペースを選択し、[データ統合へ移動] をクリックします。

    2. 左側のナビゲーションウィンドウで Synchronization Task をクリックします。ページの上部で Create Synchronization Task をクリックし、以下の基本情報を設定します。

      • [ソースと宛先]： Kafka → OSS

      • New Node Name： 同期タスクのカスタム名を入力します。

      • Synchronization Method： 単一テーブルのリアルタイム。

    2. ネットワークとリソースの設定

    1. Network and Resource Configuration セクションで、同期タスクの Resource Group を選択します。Task Resource Usage に特定の数のコンピュートユニット (CU) を割り当てることができます。

    2. Source Information で、ご利用の Kafka データソースを選択します。Destination で、ご利用の OSS データソースを選択します。その後、Test Connectivity をクリックします。image

    3. 両方の接続テストが成功したら、Next をクリックします。

    3. 同期パイプラインの設定

    1. Kafka ソースの設定

    ページの上部で Kafka ソースノードをクリックし、Source Information を設定します。

    image

    1. Source Information セクションで、同期する Kafka トピックを選択します。

      タスク作成時に生成されたデフォルト値を使用するか、必要に応じて変更できます。

    2. 右上隅にある Data Sampling をクリックします。

      表示されるダイアログボックスで、Start Time と Number of Sample Records を指定し、Start Sampling ボタンをクリックして、指定した Kafka トピックからデータをサンプリングします。トピック内のデータをプレビューすることもでき、これは後続のデータ処理ノードのデータプレビューと可視化設定の入力となります。

    3. Output Field Configuration セクションで、同期するフィールドを選択します。

    2. データ処理ノードの設定

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

    image

    データ処理ノードを設定した後、右上隅にある Output Preview をクリックできます。ダイアログボックスで Retrieve Upstream Output をクリックすると、現在のノードが Kafka トピックからサンプリングしたデータをどのように処理するかをプレビューできます。

    image

    説明

    データ出力のプレビューは、Kafka ソースの Data Sampling に依存します。出力をプレビューする前に、まず Kafka ソース設定フォームでデータをサンプリングする必要があります。

    3. OSS 送信先の設定

    ページの上部で OSS 送信先ノードをクリックし、[OSS 送信先情報] を設定します。

    image

    1. [OSS 送信先情報] セクションで、OSS にデータを書き込むための基本情報を設定します。

      • Write Format： サポートされているフォーマットは Hudi、Paimon、Iceberg です。

      • Location for Creating Metadatabase： ご利用のアカウントで Data Lake Formation (DLF) が有効になっている場合、データがデータレイクに同期される際に、システムは DLF に対応するメタデータベースとメタテーブルを自動的に作成できます。

        説明

        リージョンをまたいだメタデータベースの作成はサポートされていません。

      • Select Storage Path： 取り込み後にデータが保存される OSS 内のパスを選択します。

      • Target Database： データの送信先データベースを選択します。Create Database を選択して DLF メタデータベースを作成することもできます。その場合は、Database Name を指定する必要があります。

      • Destination Table： Create tables automatically を選択して新しいテーブルにデータを書き込むか、Use Existing Table を選択して既存のテーブルにデータを書き込むかを指定します。

      • Table Name： OSS の送信先テーブルの名前を入力または選択します。

    2. (任意) テーブル構造を編集します。

      Create tables automatically を選択した場合、Edit Table Schema ボタンをクリックして、ダイアログボックスでターゲットテーブルの構造を編集します。Regenerate Schema from Upstream Node をクリックして、アップストリームノードの出力列からテーブル構造を自動的に生成することもできます。生成された構造では、列を選択してプライマリキーとして設定できます。

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

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

    4. アラート

    タスクエラーによる業務データの同期遅延を防ぐために、同期タスクにアラートポリシーを設定できます。

    1. ページの右上隅にある Alert Settings をクリックして、Alert Rule Configurations for Real-time Synchronization Subnode ページを開きます。

    2. Add Alert Rule をクリックして、アラートルールを設定します。

      説明

      ここで定義したアラートルールは、このタスクが生成するリアルタイム同期サブタスクに適用されます。タスクを設定した後、これらのサブタスクのアラートルールは リアルタイム同期タスクの実行と管理 ページで表示および変更できます。

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

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

    5. 詳細設定

    同期タスクには、必要に応じて変更できるいくつかのパラメーターが用意されています。

    説明

    変更を加える前に、各パラメーターの機能を十分に理解し、予期せぬエラーやデータ品質の問題を防ぐようにしてください。

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

    2. [詳細設定] ページで、必要に応じてパラメーター値を変更します。

    6. DDL 機能の設定

    ソースで DDL 操作が実行される場合があります。ページの右上隅にある [DDL 機能の設定] をクリックして、ビジネス要件に基づいてソースからの DDL メッセージを処理するルールを設定できます。

    説明

    詳細については、「DDL メッセージを処理するルールの設定」をご参照ください。

    7. リソースグループの設定

    右上隅の Configure Resource Group をクリックすると、タスクの現在のリソースグループを表示および切り替えることができます。

    8. シミュレーションの実行

    タスクを設定した後、右上隅にある Dry Run をクリックします。この機能は、少量のデータサンプルでタスク全体をシミュレーションし、ターゲットテーブルでの結果をプレビューできます。設定エラー、ランタイム例外、またはダーティデータがある場合、リアルタイムでエラーメッセージが表示されます。これにより、タスクが正しく設定され、期待どおりの結果が得られることを迅速に確認できます。

    1. ダイアログボックスで、サンプリングパラメーターの Start Time と Number of Sample Records を設定します。

    2. Start Sampling をクリックして、サンプルデータを収集します。

    3. Preview をクリックして、サンプリングされたデータを使用してタスク全体の処理をシミュレーションします。

    9. 同期タスクの実行

    1. すべての設定が完了したら、ページ下部の Complete をクリックします。

    2. Data Integration > Synchronization Task ページで、作成したタスクを見つけ、Operations 列の Start をクリックします。

    3. Task List 内の対応するタスクの Name/ID をクリックして、詳細な実行プロセスを表示します。

    同期タスクの管理

    タスクステータスの表示

    同期タスクページには、すべてのタスクとその基本情報が一覧表示されます。

    image

    • [操作] 列では、同期タスクを Start または Stop できます。[その他] メニューには、Edit や View などの追加オプションがあります。

    • 実行中のタスクについては、Execution Overview を確認して、その運用状況を確認できます。概要エリアをクリックして、詳細な実行情報を表示することもできます。

    image

    Kafka から OSS への単一テーブルのリアルタイム同期タスクは、次の 2 つのステップで構成されます：

    • Schema Migration： このステップには、送信先テーブルの作成方法 (既存のテーブルの使用または自動テーブル作成) が含まれます。自動テーブル作成を選択した場合、システムはテーブル作成に使用された DDL 文を表示します。

    • Real-time Data Synchronization： このステップには、リアルタイムの運用情報、DDL レコード、アラート情報など、リアルタイムデータ同期の統計が含まれます。

    タスクの再実行

    同期されたフィールドを変更したり、ターゲットテーブル情報を調整したりする必要があるなどの特別な場合には、同期タスクの Operations 列にある Rerun をクリックできます。この操作により、調整されたフィールドやその他の変更がターゲットに同期されます。このプロセスでは、変更されていない、以前に同期されたテーブルはスキップされます。

    • 変更なしでタスクを再度実行するには、Rerun をクリックします。

    • タスクを編集した場合は、変更を加えた後に Complete をクリックします。タスクの操作は Apply Updates に変わります。Apply Updates をクリックすると、新しい設定でタスクが再実行されます。