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

使用上の注意

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

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

手順

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

1. 変更追跡タスクを作成します。 詳細については、「変更追跡シナリオの概要」の関連トピックをご参照ください。

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-SNAPSHOT</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-******.******.***:****** --topic cn_hangzhou_rm_**********_dtstest_version2 --sid dts****** --user dtstest --password Test123456 --checkpoint 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ページに移動します。

パラメーター

よくある質問

Cluster changed from *** to ***, consumer require restart.