Consume tracked data using flink-dts-connector

After you configure a change tracking instance for Data Transmission Service (DTS), you can use the flink-dts-connector to build a Flink program that consumes the tracked data.

Usage notes

The connector supports Flink programs that use the DataStream API, or the Table API and SQL.
If your Flink program uses the Table API and SQL, you can configure it to consume data from only one table at a time. To consume data from multiple tables, you must configure and run a separate task for each table.

Procedure

This topic uses IntelliJ IDEA Community Edition 2020.1 for Windows as an example to show how to use the flink-dts-connector to consume tracked data from a change tracking instance.

Create a change tracking instance. For more information, see Create a change tracking instance for an ApsaraDB RDS for MySQL instance, Create a change tracking instance for a PolarDB for MySQL cluster, or Create a change tracking instance for an ApsaraDB for Oracle instance.
Create one or more consumer groups. For more information, see Create a consumer group.
Download the flink-dts-connector file and decompress it.
Start IntelliJ IDEA and click Open or Import.
In the dialog box that appears, navigate to the directory where you decompressed the flink-dts-connector file. Expand the folders to find the Project Object Model (POM) file: pom.xml.
In the dialog box that appears, select Open as Project.

Add the following dependency to the pom.xml file:

<dependency>
      <groupId>com.alibaba.flink</groupId>
      <artifactId>flink-dts-connector</artifactId>
      <version>1.1.1-SNAPSHOT</version>
      <classifier>jar-with-dependencies</classifier>
</dependency>

In IntelliJ IDEA, expand the project folders and select the appropriate Java file based on the type of Flink API your program uses.

If your Flink program uses the DataStream API, double-click the flink-dts-connector-master\src\test\java\com\alibaba\flink\connectors\dts\datastream\DtsExample.java file and perform the following steps:

From the top menu bar, choose Run > Run....
In the pop-up window, click Edit > DtsExample.
In the Program arguments field, enter the parameters and their values as shown in the following example. Then, click Run to start the flink-dts-connector.

Note
For details about the parameters and how to find their values, see Parameters.
```
--broker-url dts-cn-******.******.***:****** --topic cn_hangzhou_rm_**********_dtstest_version2 --sid dts****** --user dtstest --password Test123456 --checkpoint 1624440043
```

The output indicates that the program is successfully tracking data changes from the source database. After startup, the data change records in the DataStream look similar to the following example.

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]}

Note

To view the details of the data change records, log on to the Task Manager UI of your Flink program.

If your Flink program uses the Table API and SQL, double-click the flink-dts-connector-master\src\test\java\com\alibaba\flink\connectors\dts\sql\DtsTableISelectTCaseTest.java file and perform the following steps:

Note

A single DtsTableISelectTCaseTest.java file supports configuring and consuming tracked data from only one table. To consume data from multiple tables, you must configure and run a separate task for each table.

Comment out the properties.load line by adding the // prefix.

/*Loads the parameter values that you set on the platform page into the Properties object.*/
//properties.load(new StringReader(new String(Files.readAllBytes(Paths.get(configFilePath)), StandardCharsets.UTF_8)));

Define the single table from which you want to consume data.

Set the parameters for the change tracking instance. For details about the parameters and how to find their values, see Parameters.

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')";
}

At the top of the IntelliJ IDEA interface, click Run'DtsTableISelectTCaseTest' to start the flink-dts-connector.

The output indicates that the program is successfully tracking data changes from the source database. After startup, the terminal displays changelog records. Update events appear in pairs: the value before the update (-U) and the value after the update (+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))

Note

To view the details of the data change records, log on to the Task Manager UI of your Flink program.

Parameters

DataStream API parameter	Table API parameter	Description	How to obtain
`broker-url`	`dts.server`	The endpoint and port of the change tracking instance. Note To minimize network latency, use the internal endpoint if your Flink program's ECS instance and the change tracking instance are in the same virtual private cloud (VPC) or classic network. We do not recommend using a public endpoint because of potential network instability.	In the Data Transmission Service (DTS) console, click the ID of the target change tracking instance. On the View Task Settings page, you can find the Topic, endpoint, and port.
`topic`	`topic`	The tracked topic of the change tracking instance.
`sid`	`dts.sid`	The ID of the consumer group.	In the DTS console, click the ID of the target change tracking instance, and then click Consume Data. You can find the ID and Account of the consumer group. Note The password for the consumer group username is the one you specified when you created the consumer group.
`user`	`dts.user`	The username for the consumer group. Warning The flink-dts-connector provided in this topic automatically handles the required username formatting. If you use a different client, you must manually format the username as `<username>-<consumer group ID>` (for example, `dtstest-dtsae******bpv`) to prevent connection failures.
`password`	`dts.password`	The password for the username.
`checkpoint`	`dts.checkpoint`	The consumer offset. This is the timestamp from which the flink-dts-connector starts consuming data. The value must be a Unix timestamp, for example, 1624440043. Note The consumer offset is useful in the following scenarios: If the consumer program is interrupted, use the last processed offset to resume consumption and prevent data loss. When the consumer program starts, provide a specific offset to start consuming data from a desired point in time.	The consumer offset must be within the data range of the change tracking instance. On the View Task Settings page of the DTS change tracking instance, you can find the start and end times in the Data Range field. You must set the consumer offset to a time within this range and convert it to a Unix timestamp. Note Use an online search engine to find a Unix timestamp converter.
N/A	`dts-cdc.table.name`	This parameter specifies the table for change tracking. Only one table is supported per task. Use the following format: For MySQL, PolarDB for MySQL, PolarDB-X 1.0, and PolarDB-X 2.0 databases, use the format `<Database name>.<Table name>`. For other database types, use the format `<Schema name>.<Table name>`.	In the DTS console, click the ID of the target change tracking instance. On the View Task Settings page, click View Objects in the upper-right corner to find the database and table names of the objects.

FAQ

Error message	Possible cause	Solution
`Cluster changed from * to *, consumer require restart.`	The DStore module, which DTS uses to read incremental data, has been switched over. This causes the consumer offset of the Flink program to be lost.	Do not restart the program. Instead, find the last known consumer offset and pass it as the `checkpoint` or `dts.checkpoint` parameter in your DtsExample.java or DtsTableISelectTCaseTest.java file to resume consumption.