8. Flink-CDC

1. Flink-CDC的介绍

Flink-cdc主要是用来同步数据库中的数据，它的主要优势在于基于Flink框架直接用Flink Stream Api 或Flink SQL 直接编程，不需要引入第三方组件

2.Flink-CDC的使用

Flink-cdc在使用上需要注意的点

• 注意Flink-cdc在2.1版本之前需要导入MySQL的连接包，之后版本不需要，如果环境中有MySQL的连接包需要去除掉
• 在2.4版本之监控MySQL表需要它有主键，2.4版本开始只需要配置“scan.incremental.snapshot.chunk.key-column”参数即可
• MySQL CDC Connector在监控多个表的时候，每个表需要指定库名，并用逗号隔开
• Flink中必须要设置checkpoint，不设置无法正常监控binlog变更日志
  Flink-CDC基于DataStream的使用

MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
        .hostname("node2")      //设置MySQL hostname
        .port(3306)             //设置MySQL port
        .databaseList("db1")    //设置捕获的数据库
        .tableList("db1.tbl1,db1.tbl2") //设置捕获的数据表
        .username("root")       //设置登录MySQL用户名
        .password("123456")     //设置登录MySQL密码
        .deserializer(new JsonDebeziumDeserializationSchema()) //设置序列化将SourceRecord 转换成 Json 字符串
        .startupOptions(StartupOptions.initial())
        .build();
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//开启checkpoint
env.enableCheckpointing(5000);
env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(),"MySQL Source")
        .setParallelism(4)
        .print();
env.execute();

基于Flink Sql的使用

EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
TableEnvironment tableEnv = TableEnvironment.create(settings);
//设置checkpoint
tableEnv.getConfig().getConfiguration().setLong("execution.checkpointing.interval", 5000L);
tableEnv.executeSql("" +
        "CREATE TABLE mysql_binlog (" +
        " id INT," +
        " name STRING," +
        " age INT," +
        " PRIMARY KEY(id) NOT ENFORCED" +
        ") WITH (" +
        " 'connector' = 'mysql-cdc'," +
        " 'hostname' = 'node2'," +
        " 'port' = '3306'," +
        " 'username' = 'root'," +
        " 'password' = '123456'," +
        " 'database-name' = 'db1'," +
        " 'table-name' = 'tbl1'" +
        ")");
tableEnv.executeSql("select * from mysql_binlog").print();

2.1 Flink-CDC对全量和增量数据的工作原理

并行读取表的全量快照，然后以单并行度方式读取表的binlog进行增量数据的同步

• 全量同步过程中，它会根据主键把数据分为多个chunk分片，然后分配给多并行度去分别读取这些chunk上的数据，读取快照期间，Flink支持chunk级别的checkpoint，即使在同步的过程中发生故障，也可以做到exactly-once级别的恢复

2.2 Flink-CDC启动模式

启动模式是指程序启动的时候，以怎么的方式监控数据库中的数据，共有如下几种模式

• initial（默认）： 对受监控的库表进行初始快照，并继续读取最新的binlog
• earliest-offset： 它会跳过快照直接读取最早的binlog日志，它与initial方式区别在于，initial只读取已经操作后（表中现有数据）的数据
• latest-offset： 不执行快照，从binlog的最新处开始读取增量数据
• specific-offset： 从指定的binlog位点开始读取，位点可以通过binlog文件名和位置指定
• timestamp： 从指定的时间戳读取binlog事件