MySQL DataStream connector - Realtime Compute for Apache Flink

This topic describes how to debug and run a DataStream job that uses the MySQL connector.

MySQL CDC DataStream API

Important

To read or write data by using the DataStream API, you must use the corresponding DataStream connector. For more information about how to configure DataStream connectors, see DataStream connector setup.

Create a DataStream API program and use MySqlSource. The following examples show the code and the pom dependencies:

Java

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;

public class MySqlSourceExample {
  public static void main(String[] args) throws Exception {
    MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
        .hostname("yourHostname")
        .port(yourPort)
        .databaseList("yourDatabaseName") // set captured database
        .tableList("yourDatabaseName.yourTableName") // set captured table
        .username("yourUsername")
        .password("yourPassword")
        .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
        .build();
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // enable checkpoint
    env.enableCheckpointing(3000);
    env
      .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
      // set 4 parallel source tasks
      .setParallelism(4)
      .print().setParallelism(1); // use parallelism 1 for sink to keep message ordering
    env.execute("Print MySQL Snapshot + Binlog");
  }
}

XML

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-core</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-base</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-common</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-api-java-bridge</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>com.alibaba.ververica</groupId>
    <artifactId>ververica-connector-mysql</artifactId>
    <version>${vvr.version}</version>
</dependency>

When you build MySqlSource, you must specify the following parameters in the code:

Parameter	Description
hostname	The IP address or hostname of the MySQL database.
port	The port number of the MySQL database service.
databaseList	The name of the MySQL database. Note Database names support regular expressions to read data from multiple databases. You can use `.*` to match all databases.
username	The username of the MySQL database service.
password	The password of the MySQL database service.
deserializer	The deserializer that converts records of the SourceRecord type to a specified type. Valid values: RowDataDebeziumDeserializeSchema: converts SourceRecord to RowData, the internal data structure of Flink Table or SQL. JsonDebeziumDeserializationSchema: converts SourceRecord to a JSON string.

You must specify the following parameters in the pom dependencies:

${vvr.version}	The engine version of Realtime Compute for Apache Flink. Example: `1.17-vvr-8.0.4-3`. Note Use the version number displayed on Maven as the authoritative one, because hotfix versions are released from time to time and these updates may not be announced through other channels.
${flink.version}	The Apache Flink version. Example: `1.17.2`. Important Use the Apache Flink version that corresponds to the engine version of Realtime Compute for Apache Flink to avoid incompatibility issues at runtime. For more information about the version mappings, see Engine.

DataStream debugging solution

Create a DataStream API program and use MySqlSource. The following example shows the code:

Important

For local debugging, you must download the required JAR files and configure the dependencies. For more information, see Run and debug jobs that contain connectors locally. This topic provides a sample Maven project. For more information, see MysqlCDCDemo.zip.

package com.alibaba.realtimecompute;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import org.apache.flink.configuration.Configuration;

public class MysqlCDCDemo {

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setString("pipeline.classpaths", "file://" + "absolute path to the MySQL uber JAR");  // Configure the dependency
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hostname")
                .port(3306)
                .databaseList("test_db") // Set the database to capture
                .tableList("test_db.test_table") // Set the table to capture
                .username("username")
                .password("password")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();
        env.enableCheckpointing(3000);
        env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
                .setParallelism(4)
                .print().setParallelism(1);
        env.execute("Print MySQL Snapshot + Binlog");
    }
}

When you build MySqlSource, you must specify the following parameters in the code:

Parameter	Description
hostname	The IP address or hostname of the MySQL database.
port	The port number of the MySQL database service.
databaseList	The name of the MySQL database. Note Database names support regular expressions to read data from multiple databases. You can use `.*` to match all databases.
username	The username of the MySQL database service.
password	The password of the MySQL database service.
deserializer	The deserializer that converts records of the SourceRecord type to a specified type. Valid values: RowDataDebeziumDeserializeSchema: converts SourceRecord to RowData, the internal data structure of Flink Table or SQL. JsonDebeziumDeserializationSchema: converts SourceRecord to a JSON string.

pom dependencies

Local debugging

Realtime Compute for Apache Flink connectors contain commercialized additional content and differ from Apache Flink in many ways. For local debugging, make the following changes to the pom file:

${flink.version} must be 1.19.0.
Recommended versions of the connector ${vvr.version}:
1. For the VVR 8.x engine, use version 1.17-vvr-8.0.11-4.
2. For the VVR 11.x engine, use the latest connector version that corresponds to your engine version. For example, for the VVR 11.8 engine, use the latest version 1.20-vvr-11.8.0-1-jdk11. For other versions, see Maven Repository.
You must add the Kafka connector dependency.

<properties>
    <maven.compiler.source>8</maven.compiler.source>
    <maven.compiler.target>8</maven.compiler.target>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <java.version>8</java.version>
    <flink.version>1.19.0</flink.version>
    <vvr.version>1.20-vvr-11.8.0-1-jdk11</vvr.version>
</properties>
<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-core</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <!-- Use this group ID for VVR 11.x -->
        <groupId>com.alibaba.ververica</groupId>

       <!-- Use this group ID for VVR 8.x -->
        <!-- <groupId>org.apache.flink</groupId> -->
        <artifactId>flink-table-common</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>com.alibaba.ververica</groupId>
        <artifactId>ververica-connector-mysql</artifactId>
        <version>${vvr.version}</version>
    </dependency>
    <dependency>
        <groupId>com.alibaba.ververica</groupId>
        <artifactId>ververica-connector-kafka</artifactId>
        <version>${vvr.version}</version>
    </dependency>
</dependencies>

Deployment and debugging on Realtime Compute for Apache Flink

When you debug a job on Realtime Compute for Apache Flink, no connector version limits apply. Make sure that ${flink.version} and ${vvr.version} correspond to the engine version of the job. For more information about the mappings, see Engine. The following pom file is for reference:

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-core</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-base</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-common</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-api-java-bridge</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>com.alibaba.ververica</groupId>
    <artifactId>ververica-connector-mysql</artifactId>
    <version>${vvr.version}</version>
</dependency>