欢迎光临
flinkcdc 原理 + 实践
   

flinkcdc 原理 + 实践

使用环境

        Flink 1.14.2 + flink cdc 2.2.0

        提示:flinkcdc 2.2版本之后才支持flink 1.14.*,

                  flinkcdc 2.2版本之前不支持 mysql低版本5.6的 cdc.      

CDC1.*版本痛点

  1. 单并发。

  2. 为了保证一致性,一般通过全量 + 增量进行获取数据。 在全量阶段会进行加锁操作。 通过对 binlog的起始位置和 表的 schema 锁住,保证在全量读取的时候阻止所有新的update。

  3. 全量获取数据的时候不支持checkpoint, fail后需要重新读取。

CDC2.*版本介绍

flinkCDC2.* 优点 :

  并发读取全量数据后无缝转换为单线程读取增量

   通过高水位的方式来替代1.*版本的加锁操作来保证数据一致性。解决1.*版本痛点。

  • chunk切分:

    SourceEnumerator 将表按主键切分chunk.(没有主键可以指定) 分片数表示每个chunk用对应的task并行执 行。达到多并发。 task对 chunk 一对多的关系。 也就是说 一个sourceReader对应一个或多个chunk.

  • chunk分配:

    SourceReader , 每个SourceReader读取表中的一部分数据达到并行读取。

  • chunk读取: (单个Chunk中的数据一致性)

    SourceReader在读取表数据之前,也就是在读取这个chunk切片的时候,会记录当前的binlog位置信息为 低位点。 每个 chunk获取的低位点可能都不一样:比如: 10,30,20 的低位点。

    SourceReader将自身区间内的数据查询并放入buffer(快照读取,等待修正)

    SourceReader查询完成该chunk切片数据之后记录当前binlog位置信息为 高位点。 每个chunk获取的高位点可能也都不一样。比如: 80,70,90 的高位点。

    全量到增量衔接部分:每个chunk消费从低位点到高位点之间的binlog。对buffer数据进行修正。chunk最终的输出则是在对应的该高位点最新的数据。 但是目前只保证了单个chunk中的数据一致性。

  • chunk汇报:

    chunk读取完成之后,会对SourceEnumerator进行汇报,为的是后续的分发binlog chunk. 因为之前的数据针对每个chunk对应的高位点或者低位点获取到的数据范围不一致,为了保证后续继续处理增量的binlog,需要进行分发binlog.

  • chunk分配: (多个Chunk的一致性, 增量阶段)

    在后续需要处理的增量的binlog中,SourceEnumerator会通过下发binlog chunk给任意一个SourceReader进行单并发增量实现。因为增量通常都是数据库单并行操作的,多个task没有意义

    至此切换至增量阶段,会从已完成的全量的所有chunk中筛选出最小的binglog hw。 也就是70开始,当数据到来,判断数据所属chunk,其次,根据状态判断当前数据偏移量如果大于快照时的偏移量 70,那么进行下发,如果小于,则是该chunk的重复数据,直接丢弃。

    解决了全量时期单并发的痛点。并通过高水位方式保证了数据一致性,无缝切换全量 + 增量 阶段。

flink cdc datastream + flink cdc dql 使用

final MySqlSource build = MySqlSource.builder()
        .hostname("127.0.0.1")
        .port(3307)
        .username("root")
        .password("123456")
        .databaseList("coocaa")
        .tableList("coocaa.Course")
        .deserializer(new StringDebeziumDeserializationSchema())
        .startupOptions(StartupOptions.initial())
        // initial:  表示先将之前数据同步,在根据现有的binlog持续更新。
        .build();
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
final DataStreamSource dataStreamSource = env.fromSource(build, WatermarkStrategy.noWatermarks(),"mysql-cdc");
dataStreamSource.print();
env.execute();
// TODO: 初始化flink
StreamExecutionEnvironment streamEnv
        = StreamExecutionEnvironment.getExecutionEnvironment();
EnvironmentSettings envSettings = EnvironmentSettings.newInstance().inStreamingMode().build();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(streamEnv, envSettings);
tableEnv.executeSql("" +
        "create table tablea(                                   \n" +
        "       database_name STRING METADATA VIRTUAL,          \n" +
        "       table_name STRING METADATA VIRTUAL,             \n" +
        "       cloumn1         String,                         \n" +
        "       cloumn2         String,                         \n" +
        "       cloumn3         String,                         \n" +
        "       cloumn4         String,                         \n" +
        "       cloumn5         String,                         \n" +
        "       cloumn6         String,                         \n" +
        "       cloumn7         String,                         \n" +
        "       cloumn8         int,                            \n" +
        "       cloumn9         int,                            \n" +
        "       cloumn10        int,                            \n" +
        "       cloumn11        String,                         \n" +
        "        PRIMARY KEY (cloumn1) NOT ENFORCED             \n" +
        " ) with (                                              \n" +
        "      'connector' = 'mysql-cdc',                       \n" +
        "      'hostname' = '***',                              \n" +
        "      'port' = '***',                                  \n" +
        "      'database-name' = '***',                         \n" +
        "      'username' = 'root',                             \n" +
        "      'password' = '123456',                           \n" +
        "      'table-name' = '***.*',                          \n" +   //可通过正则进行多库多表
        "      'scan.incremental.snapshot.enabled' = 'false',   \n" +
        "      'scan.startup.mode' = 'initial'                  \n" +       //initial  or latest-offset
        " )");
tableEnv.from("tablea").printSchema();
tableEnv.executeSql("select * from tablea").print();
 
 美人梅树苗  桂枝香王安石  摇钱树怎么种表情包  如意杨幂  家里最旺财的十种花  纤夫的爱谁唱的 
打赏
版权声明:本文采用知识共享 署名4.0国际许可协议 [BY-NC-SA] 进行授权
文章名称:《flinkcdc 原理 + 实践》
文章链接:https://goodmancom.com/wl/175748.html