Flink-CDC-快速入手（MySQL为例）_flinkcdc mysql 连接断

0. 简介

官方文档地址

FlinkCDC是什么

• CDC Connectors for Apache Flink 是一组用于Apache Flink 的源连接器，使用变更数据捕获(CDC)从不同的数据库接收变更。用于ApacheFlink 的CDC连接器集成了Debezium（讲数据库转换成事件流）作为捕获数据更改的引擎。因此，它可以充分利用Debezium的能力
• 将数据库中的数据转换成Flink的数据流（StreamData）或 Table&SQL 的形式进行处理操作

支持的连接源及版本

FlinkCDC与Flink版本

1. 用法（MySQL为例）

Flink-CDC-MySQL文档地址

DataStream Source

代码

public static void main(String[] args) throws Exception {

    //1.创建执行环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    //2.Flink-CDC 将读取 binlog 的位置信息以状态的方式保存在 CK,如果想要做到断点续传,需要从 Checkpoint 或者 Savepoint 启动程序
    //2.1 开启 Checkpoint,每隔 5 秒钟做一次 CK
    env.enableCheckpointing(5000L);
    //2.2 指定 CK 的一致性语义
    env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
    //2.3 设置任务关闭的时候保留最后一次 CK 数据
    env.getCheckpointConfig().enableExternalizedCheckpoints(
        CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
    //2.4 指定从 CK 自动重启策略
    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 2000L));
    //2.5 设置状态后端
    env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/flinkCDC"));
    //2.6 设置访问 HDFS 的用户名
    System.setProperty("HADOOP_USER_NAME", "xxx");

    //3.创建 Flink-MySQL-CDC 的 Source
    DebeziumSourceFunction<String> mysqlSource = MySqlSource.<String>builder()
        .hostname("yourHostname")
        .port(yourPort)
        .username("yourUsername")
        .password("yourPassword")
        .databaseList("test")
        .tableList("test.book")  //可选配置项,如果不指定该参数,则会读取databaseList下的所有表的数据，注意：指定的时候需要使用"db.table"的方式
        .deserializer(new StringDebeziumDeserializationSchema())  // 序列化，输出的数据格式，StringDebeziumDeserializationSchema是String格式的数据
        .build();

    //4.使用 CDC Source 从 MySQL 读取数据
    DataStreamSource<String> sourceData = env.addSource(mysqlSource);

    //5.打印数据
    sourceData.print();

    //6.执行任务
    env.execute();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

自定义反序列化器

• 上面所用的反序列化器new StringDebeziumDeserializationSchema()是自带的String字符串序列化器

• 我们可以自定义反序列化器，实现 DebeziumDeserializationSchema

  public static void main(String[] args) throws Exception {
  
      //1.创建执行环境
      StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  
      //2.Flink-CDC 将读取 binlog 的位置信息以状态的方式保存在 CK,如果想要做到断点续传,需要从 Checkpoint 或者 Savepoint 启动程序
      //2.1 开启 Checkpoint,每隔 5 秒钟做一次 CK
      env.enableCheckpointing(5000L);
      //2.2 指定 CK 的一致性语义
      env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
      //2.3 设置任务关闭的时候保留最后一次 CK 数据
      env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
      //2.4 指定从 CK 自动重启策略
      env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 2000L));
      //2.5 设置状态后端
      env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/flinkCDC"));
      //2.6 设置访问 HDFS 的用户名
      System.setProperty("HADOOP_USER_NAME", "xxx");
  
      //3.创建 Flink-MySQL-CDC 的 Source
      DebeziumSourceFunction<String> mysqlSource = MySqlSource.<String>builder()
          .hostname("yourHostname")
          .port(yourPort)
          .username("yourUsername")
          .password("yourPassword")
          .databaseList("test")
          .tableList("test.book")
          .deserializer(new DebeziumDeserializationSchema<String>() {
              /**
                       * 反序列化操作
                       * @param sourceRecord  源数据
                       * @param collector  
                       * @throws Exception
                       */
              @Override
              public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
                  // TODO 序列化逻辑，最终将反序列化后的数据通过 collector.collect() 方法输出
                  collector.collect(xxx);
              }
  
              // 序列化类型
              @Override
              public TypeInformation<String> getProducedType() {
                  return TypeInformation.of(String.class);
              }
          })
          .build();
  
      //4.使用 CDC Source 从 MySQL 读取数据
      DataStreamSource<String> sourceData = env.addSource(mysqlSource);
  
      //5.打印数据
      sourceData.print();
  
      //6.执行任务
      env.execute();
  
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19
  20
  21
  22
  23
  24
  25
  26
  27
  28
  29
  30
  31
  32
  33
  34
  35
  36
  37
  38
  39
  40
  41
  42
  43
  44
  45
  46
  47
  48
  49
  50
  51
  52
  53
  54
  55
  56
  57
  58

Table/SQL API

代码

public static void main(String[] args) throws Exception {

    //1.创建执行环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

    //2.创建 Flink-MySQL-CDC 的 Source
    tableEnv.executeSql("CREATE TABLE user_info (" +
                        " id INT," +
                        " name STRING," +
                        " phone_num STRING" +
                        ") WITH (" +
                        " 'connector' = 'mysql-cdc'," +
                        " 'hostname' = 'yourHostname'," +
                        " 'port' = 'yourPort'," +
                        " 'username' = 'yourUsername'," +
                        " 'password' = 'yourPassword'," +
                        " 'database-name' = 'gmall-flink'," +
                        " 'table-name' = 'z_user_info'" +
                        ")");

    // 3.执行sql，并打印
    tableEnv.executeSql("select * from user_info").print();

    // 4.执行任务
    env.execute();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

MySQL参数列举

Maven

<properties>
    <flink.version>1.13.0</flink.version>
    <java.version>1.8</java.version>
    <scala.binary.version>2.12</scala.binary.version>
    <slf4j.version>1.7.30</slf4j.version>
</properties>




<dependency>
    <groupId>com.ververica</groupId>
    <artifactId>flink-connector-mysql-cdc</artifactId>
    <version>2.0.0</version>
</dependency>


<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
</dependency>

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
</dependency>

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
</dependency>

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-common</artifactId>
    <version>${flink.version}</version>
</dependency>

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-json</artifactId>
    <version>${flink.version}</version>
</dependency>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46