Realtime Compute for Apache Flink:MySQL CDC DataStream connector

Each MySQL database client that is used to read data from binary data files must have a unique server ID. The MySQL server maintains network connections and binary log file positions based on the server ID. Therefore, if different deployments share the same server ID, data may be read from an invalid binary log file position. In this case, data loss, data duplication, or a surge in CPU utilization of the MySQL server occurs. This affects the stability of online business. To address these issues, we recommend that you explicitly configure a unique server ID for each MySQL data source of each DataStream deployment. The following sample code shows how to configure server IDs.

MySqlSource.builder().serverId("56000")

The Ververica Runtime (VVR) connector is stored in the Maven central repository for you to use during draft development. If you want to use the MySQL CDC DataStream connector, perform the following steps:

1. Step 1: Prepare the development environment for a DataStream draft

2. Step 2: Develop a DataStream draft

3. Step 3: Package the program and publish the DataStream draft

We recommend that you package the connector as a project dependency into the JAR file of your draft. Perform the following steps:

1. Add the following configurations to the POM file of the Maven project to reference SNAPSHOT repositories:

   <repositories>
     <repository>
       <id>oss.sonatype.org-snapshot</id>
       <name>OSS Sonatype Snapshot Repository</name>
       <url>http://oss.sonatype.org/content/repositories/snapshots</url>
       <releases>
         <enabled>false</enabled>
       </releases>
       <snapshots>
         <enabled>true</enabled>
       </snapshots>
     </repository>
     <repository>
       <id>apache.snapshots</id>
       <name>Apache Development Snapshot Repository</name>
       <url>https://repository.apache.org/content/repositories/snapshots/</url>
       <releases>
         <enabled>false</enabled>
       </releases>
       <snapshots>
         <enabled>true</enabled>
       </snapshots>
     </repository>
   </repositories>

2. Check whether your settings.xml configuration file contains the <mirrorOf>*</mirrorOf> configuration.

   If the <mirrorOf>*</mirrorOf> configuration is contained in the configuration file, the current mirror contains all repositories and the Maven project does not download dependencies from the preceding two specified SNAPSHOT repositories. As a result, the Maven project cannot download SNAPSHOT dependencies from these repositories. To avoid this issue, perform the following operations based on the actual scenario: If the <mirrorOf>*</mirrorOf> configuration is contained in the configuration file, change the configuration to <mirrorOf>*,!oss.sonatype.org-snapshot,!apache.snapshots</mirrorOf>. If the <mirrorOf>external:*</mirrorOf> configuration is contained in the configuration file, change the configuration to <mirrorOf>external:*,!oss.sonatype.org-snapshot,!apache.snapshots</mirrorOf>. If the <mirrorOf>external:http:*</mirrorOf> configuration is contained in the configuration file, change the configuration to <mirrorOf>external:http:*,!oss.sonatype.org-snapshot,!apache.snapshots</mirrorOf>.

3. Add the connector that you want to use to the Maven POM file as a project dependency.

   <dependency>
       <groupId>com.ververica</groupId>
       <artifactId>flink-connector-mysql-cdc</artifactId>
       <! -- vvr-4.0.X-flink-1.13 corresponds to V2.2.1 of the MySQL CDC DataStream connector  -->
       <!-- vvr-6.0.X-flink-1.15 corresponds to V2.3.0 of the MySQL CDC DataStream connector   -->
       <version>2.2.1</version>
   </dependency>

   Different connector versions may correspond to different connector types. We recommend that you use the latest version for the type of the connector that you use.

   Important

   • You must search for the connector versions that contain the SNAPSHOT keyword in the SNAPSHOT repository oss.sonatype.org. You cannot find the versions in the Maven central repository search.maven.org.

   • If you use multiple connectors, you must merge the files in the META-INF directory. To merge the files, add the following code to the POM file:

     <transformers>
         <!-- The service transformer is needed to merge META-INF/services files -->
         <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
         <transformer implementation="org.apache.maven.plugins.shade.resource.ApacheNoticeResourceTransformer">
             <projectName>Apache Flink</projectName>
             <encoding>UTF-8</encoding>
         </transformer>
     </transformers>

Create a DataStream API program and use the MySqlSource class. Sample statements:

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

The following table describes the parameters that you must configure for the MySqlSource class.