AI 助理
文档备案控制台
官方文档
首页 实时计算 Flink版 流计算Flink 开发参考 连接器 MySQL MySQL DataStream连接器

MySQL DataStream连接器

更新时间：
复制 MD 格式
产品详情
我的收藏

本文为您介绍如何调试和运行使用MySQL连接器的 DataStream 作业。

MySQL CDC DataStream API

重要

通过DataStream的方式读写数据时，则需要使用对应的DataStream连接器连接Flink，DataStream连接器设置方法请参见DataStream连接器使用方法。

创建DataStream API程序并使用MySqlSource。代码及pom依赖项示例如下：

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

在构建MySqlSource时，代码中必须指定以下参数：

参数

说明

hostname

MySQL数据库的IP地址或者Hostname。

port

MySQL数据库服务的端口号。

databaseList

MySQL数据库名称。

说明

数据库名称支持正则表达式以读取多个数据库的数据，您可以使用.*匹配所有数据库。

username

MySQL数据库服务的用户名。

password

MySQL数据库服务的密码。

deserializer

反序列化器，将SourceRecord类型记录反序列化到指定类型。参数取值如下：

  • RowDataDebeziumDeserializeSchema：将SourceRecord转成Flink Table或SQL内部数据结构RowData。

  • JsonDebeziumDeserializationSchema：将SourceRecord转成JSON格式的String。

pom依赖项必须指定以下参数：

${vvr.version}

阿里云实时计算Flink版的引擎版本，例如：1.17-vvr-8.0.4-3。

说明

请以Maven上显示的版本号为准，因为我们会不定期发布Hotfix版本，而这些更新可能不会通过其他渠道通知。

${flink.version}

Apache Flink版本，例如：1.17.2。

重要

请使用阿里云实时计算Flink版的引擎版本对应的Apache Flink版本，避免在作业运行时出现不兼容的问题。版本对应关系详情，请参见引擎。

DataStream调试方案

创建DataStream API程序并使用MySqlSource。代码示例如下：

重要

本地调试需要下载相关jar并配置依赖，详情请参见本地运行和调试包含连接器的作业。本文已提供Maven示例项目，详情请参考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://" + "mysql uber jar绝对路径");  // 配置依赖
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hostname")
                .port(3306)
                .databaseList("test_db") // 设置捕获的数据库
                .tableList("test_db.test_table") // 设置捕获的表
                .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");
    }
}

在构建MySqlSource时，代码中必须指定以下参数：

参数

说明

hostname

MySQL数据库的IP地址或者主机名。

port

MySQL数据库服务的端口号。

databaseList

MySQL数据库名称。

说明

数据库名称支持正则表达式以读取多个数据库的数据，您可以使用.*匹配所有数据库。

username

MySQL数据库服务的用户名。

password

MySQL数据库服务的密码。

deserializer

反序列化器，将SourceRecord类型记录反序列化到指定类型。参数取值如下：

  • RowDataDebeziumDeserializeSchema：将SourceRecord转成Flink Table或SQL内部数据结构RowData。

  • JsonDebeziumDeserializationSchema：将SourceRecord转成JSON格式的String。

pom依赖

本地调试

阿里云实时计算Flink版连接器包含商业化的额外内容，与Apache Flink版本有许多差异，本地调试时需对pom做如下修改：

  1. ${flink.version}必须为1.19.0。

  2. 连接器${vvr.version}推荐版本如下：

    1. 使用vvr 8.x版本引擎时，推荐1.17-vvr-8.0.11-4版本。

    2. 使用vvr 11.x版本引擎时，推荐使用最新的对应引擎版本的连接器，例如如果使用vvr 11.8版本引擎，推荐使用最新版本1.20-vvr-11.8.0-1-jdk11。其他版本可见Maven Repository。

  3. 需要添加kafka连接器依赖。

<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>
        <!-- VVR 11.x 版本使用此 Group ID -->
        <groupId>com.alibaba.ververica</groupId>

       <!-- VVR 8.x 版本使用此 Group ID -->
        <!-- <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>

实时计算Flink版部署调试

在实时计算Flink版启动作业调试时，连接器无版本限制，${flink.version}和${vvr.version}仅需和作业引擎版本相对应即可，对应关系请参见引擎。参考pom如下：

<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>
上一篇：MySQL 数据摄入YAML连接器下一篇：实时数仓Hologres