Data Transmission Service:flink-dts-connectorを使用した追跡データの消費

制限事項

• Data Transmission Service (DTS) は、次のタイプのFlinkプログラムをサポートしています。DataStream APIおよびTable API & SQL。

• Table APIとSQLプログラムを使用する場合、変更追跡タスクを設定するたびに1つのテーブルのデータのみを使用できます。 複数のテーブルのデータを使用する場合は、テーブルごとにタスクを設定する必要があります。

手順

この例では、Windows用IntelliJ IDEA Community Edition 2020.1が使用されています。

1. 変更追跡タスクを作成します。 詳細については、「ApsaraDB RDS For MySQLインスタンスからのデータ変更の追跡」、「PolarDB for MySQLクラスターからのデータ変更の追跡」、または「自己管理型Oracleデータベースからのデータ変更の追跡」をご参照ください。

2. 1つ以上のコンシューマグループを作成します。 詳細については、「コンシューマーグループの作成」をご参照ください。

3. flink-dts-connectorファイルをダウンロードして解凍します。

4. IntelliJ IDEAを開きます。 表示されるウィンドウで、[開く] または [インポート] をクリックします。

   

5. 表示されるダイアログボックスで、flink-dts-connectorファイルが解凍されているディレクトリに移動し、フォルダを展開してpom.xmlファイルを見つけます。

   

6. 表示されるダイアログボックスで、プロジェクトとして開く.

7. 次の依存関係をpom.xmlファイルに追加します。

   <dependency>
         <groupId>com.alibaba.flink</groupId>
         <artifactId>flink-dts-connector</artifactId>
         <version>1.1.1-スナップショット </version>
         <classifier>jar-with-dependencies</classifier>
   </dependency> 

8. IntelliJ IDEAで、フォルダを展開してJavaファイルを見つけます。 次に、使用するFlinkコネクタのタイプに基づいてJavaファイルをダブルクリックします。

   • DataStream APIコネクタを使用する場合は、flink-dts-connector-master\src\test\java\com\alibaba\flink\connectors\dts\datastream\DtsExample.javaファイルをダブルクリックし、次の操作を実行する必要があります。

     1. IntelliJ IDEAの上部メニューバーで、実行アイコンをクリックします。

     2. 表示されるダイアログボックスで、[DtsExample] > [Edit] を選択します。

     3. [プログラム引数] フィールドにパラメーターと対応する値を入力し、[実行] をクリックしてflink-dts-connectorを実行します。

        説明

        パラメーターとパラメーター値を取得するためのメソッドの詳細については、このトピックの「パラメーター」をご参照ください。

        -- broker-url dts-cn-*******. ******.***: *******-トピックcn_hangzhou_rm_*********_dtstest_version2-sid dts ********-ユーザーdtstest-パスワードTest123456-チェックポイント1624440043

     4. 次の図は、Flinkプログラムがソースデータベースからのデータ変更を追跡できることを示しています。

        説明

        データ変更の特定のレコードを照会するには、FlinkプログラムのTask Managerページに移動します。

   • Table APIとSQLコネクタを使用する場合は、flink-dts-connector-master\src\test\java\com\alibaba\flink\connectors\dts\sql\DtsTableISelectTCaseTest.javaファイルをダブルクリックし、次の操作を実行する必要があります。

     説明

     単一のDtsTableISelectTCaseTest.javaファイルを使用して、1つの変更追跡タスクのみを構成し、1つのテーブルのデータのみを使用できます。 複数のテーブルのデータを使用する場合は、テーブルごとにタスクを設定する必要があります。

     1. 次の図に示すように、スラッシュ (//) を2つ入力し、コメントを追加します。

     2. データの変更を追跡するテーブルに関する情報を指定します。 SQL文がサポートされています。

     3. 変更追跡インスタンスに必要なパラメーターを設定します。 詳細については、このトピックの「パラメーター」セクションをご参照ください。

     4. IntelliJ IDEAの上部メニューバーで、[Run'DtsTableISelectTCaseTest'] をクリックしてflink-dts-connectorを実行します。

     5. 次の図は、Flinkプログラムがソースデータベースからのデータ変更を追跡できることを示しています。

        説明

        データ変更の特定のレコードを照会するには、FlinkプログラムのTask Managerページに移動します。

パラメーター

よくある質問

クラスターが *** から *** に変更され、消費者は再起動が必要です。