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

注意事項

• このコネクタは、DataStream API、または Table API と SQL を使用する Flink プログラムをサポートします。

• Flink プログラムで Table API と SQL を使用する場合、同時に 1 つのテーブルからのみデータを消費するように設定できます。複数のテーブルからデータを消費するには、テーブルごとに個別のタスクを設定して実行する必要があります。

手順

このトピックでは、Windows 版 IntelliJ IDEA Community Edition 2020.1 を例に、flink-dts-connector を使用して変更追跡インスタンスから追跡データを消費する方法を示します。

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

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

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

4. IntelliJ IDEA を起動し、[Open or Import] をクリックします。

5. 表示されたダイアログボックスで、flink-dts-connector ファイルを解凍したディレクトリに移動します。フォルダーを展開して、Project Object Model (POM) ファイル 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 でプロジェクトフォルダーを展開し、プログラムで使用する Flink API の種類に応じて、適切な Java ファイルを選択します。

   • Flink プログラムで DataStream API を使用する場合は、flink-dts-connector-master\src\test\java\com\alibaba\flink\connectors\dts\datastream\DtsExample.java ファイルをダブルクリックし、次の手順を実行します：

     1. 上部のメニューバーから、[Run] > [Run...] を選択します。

     2. ポップアップウィンドウで、[DtsExample] > [編集]をクリックします。

     3. [Program arguments] フィールドに、次の例のようにパラメーターとその値を入力します。次に、[Run] をクリックして flink-dts-connector を起動します。

        説明

        パラメーターの詳細と値の確認方法については、「パラメーター」をご参照ください。

        --broker-url dts-cn-******.******.***:****** --topic cn_hangzhou_rm_**********_dtstest_version2 --sid dts****** --user dtstest --password Test123456 --checkpoint 1624440043

     4. 出力は、プログラムがソースデータベースのデータ変更を正常に追跡していることを示します。起動後、DataStream 内のデータ変更レコードは次の例のようになります。

        LazyParseRecord {operationType [HEARTBEAT], checkpoint [0@12006303@289540@2688@1625045211000]}
        LazyParseRecord {operationType [HEARTBEAT], checkpoint [0@12006307@290047@2688@1625045212000]}
        LazyParseRecord {operationType [UPDATE], checkpoint [0@12006305@290016@2688@1625045212000]}
        LazyParseRecord {operationType [HEARTBEAT], checkpoint [0@12006308@290047@2688@1625045214000]}
        LazyParseRecord {operationType [HEARTBEAT], checkpoint [0@12006309@290047@2688@1625045215000]}
        LazyParseRecord {operationType [HEARTBEAT], checkpoint [0@12006310@290047@2688@1625045216000]}
        LazyParseRecord {operationType [HEARTBEAT], checkpoint [0@12006311@290047@2688@1625045217000]}

        説明

        データ変更レコードの詳細を表示するには、Flink プログラムの TaskManager UI にログインします。

   • Flink プログラムで Table API と SQL を使用する場合は、flink-dts-connector-master\src\test\java\com\alibaba\flink\connectors\dts\sql\DtsTableISelectTCaseTest.java ファイルをダブルクリックし、次の手順を実行します：

     説明

     DtsTableISelectTCaseTest.java ファイルでは、1 つのテーブルからのみ追跡データを設定して消費できます。複数のテーブルからデータを消費するには、テーブルごとに個別のタスクを設定して実行する必要があります。

     1. // プレフィックスを追加して、properties.load 行をコメントアウトします。

        /* プラットフォームで設定したパラメーター値を Properties オブジェクトにロードします。 */
        //properties.load(new StringReader(new String(Files.readAllBytes(Paths.get(configFilePath)), StandardCharsets.UTF_8)));

     2. データを消費するテーブルを 1 つ定義します。

     3. 変更追跡インスタンスのパラメーターを設定します。パラメーターの詳細と値の確認方法については、「パラメーター」をご参照ください。

        public static void main(String[] args) throws Exception {
            setup(args);
            final String createTable =
                    "create table `dts` (\n"
                    + " `ts` TIMESTAMP(3) METADATA FROM 'timestamp',\n"
                    + " `id` bigint,\n"
                    + " `name` varchar,\n"
                    + " `age` bigint,\n"
                    + " WATERMARK FOR ts AS ts - INTERVAL '5' SECOND"
                    + ") with (\n"
                    + "'connector' = 'dts',"
                    + "'dts.server' = 'dts-cn-xxx:18001',"
                    + "'topic' = 'cn_hangzhou_rm_xxx_dtstest_version2',"
                    + "'dts.sid' = 'dtsxxx', "
                    + "'dts.user' = 'dtstest', "
                    + "'dts.password' = 'xxx',"
                    + "'dts.checkpoint' = '1624440043', "
                    + "'dts-cdc.table.name' = 'dtstestdata.order',"
                    + "'format' = 'dts-cdc')";
        }

     4. IntelliJ IDEA の上部で、[Run'DtsTableISelectTCaseTest'] をクリックして flink-dts-connector を起動します。

     5. 出力は、プログラムがソースデータベースのデータ変更を正常に追跡していることを示します。起動後、ターミナルに変更ログレコードが表示されます。更新イベントはペアで表示されます。更新前の値 (-U) と更新後の値 (+U) です。

        ######> (false,-U(2021-06-23T20:32:17.391,null,null,null))
        ######> (true,+U(2021-06-23T20:32:17.391,null,null,null))
        ######> (false,-U(2021-06-23T20:32:45.604,null,null,null))
        ######> (true,+U(2021-06-23T20:32:45.604,null,null,null))
        ######> (false,-U(2021-06-30T17:26:52.201,null,null,null))
        ######> (true,+U(2021-06-30T17:26:52.201,null,null,null))
        ######> (false,-U(2021-06-30T19:19:26.975,null,null,null))
        ######> (true,+U(2021-06-30T19:19:26.975,null,null,null))

        説明

        データ変更レコードの詳細を表示するには、Flink プログラムの TaskManager UI にログインします。

パラメーター

よくある質問

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