Flink CDC使用(数据采集CDC方案比较)

2022-07-10 15264

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

本文涉及的产品

实时计算 Flink 版，5000CU*H 3个月

云数据库 RDS MySQL，集群系列 2核4GB

RDS MySQL Serverless 基础系列，0.5-2RCU 50GB

简介： cdc 使用场景场景cdc方案对比flink cdc 的基本使用

什么是CDC

CDC是Change Data Capture（变更数据捕获）的简称。其核⼼原理是监测并捕获数据库的变动（增删改等），将这些变更按发⽣的顺序捕获，当然也可以写⼊到消息队列中供其他服务消费

cdc使用场景

cdc 实现

实现CDC即捕获数据库的变更数据有两种机制：

比较项	基于查询实现CDC	基于日志实现CDC
典型产品	Sqoop、DataX等	Canal、Debezium等
执⾏模式	批处理	流处理
捕获所有数据变化	NO	YES
低延迟	NO	YES
不增加数据库负载	NO	YES
不侵⼊业务(不需要lastUpdate字段)	NO	YES
捕获删除事件	NO	YES
捕获旧记录的状态	NO	YES

Flink CDC

Flink CDC是Flink社区开发的flink-cdc-connectors 组件，这是⼀个可以直接从 MySQL、PostgreSQL 等数据库直接读取全量数据和增量变更数据的 source 组件。⽬前也已开源，开源地址：

https://github.com/ververica/flink-cdc-connectors

https://ververica.github.io/flink-cdc-connectors/master/

⽀持的连接器

支持Flink 版本

Flink CDC 优势

传统的cdc不足：

传统的基于 CDC 的 ETL 分析中，数据采集⼯具是必须的，国外⽤户常⽤ Debezium，国内⽤户常⽤阿⾥开源的 Canal，采集⼯具负责采集数据库的增量数据，⼀些采集⼯具也⽀持同步全量数据。采集到的数据⼀般输出到消息中间件如 Kafka，然后 Flink 计算引擎再去消费这⼀部分数据写⼊到⽬的端，⽬的端可以是各种 DB，数据湖，实时数仓和离线数仓。

注意，Flink 提供了 changelog-json format，可以将 changelog 数据写⼊离线数仓如 Hive / HDFS；对于实时数仓，Flink ⽀持将 changelog 通过 upsert-kafka connector 直接写⼊ Kafka。

Flink CDC的基本理念就是去替换上图中红色线框内的采集组件和消息队列，从⽽简化传输链路，降低维护成本。同时更少的组件也意味着数据时效性能够进⼀步提⾼。

Flink cdc采集方案

基于FlinkCDC，我们只需要通过⼀个 Flink SQL 作业就完成了 CDC 的数据采集，加⼯和同步，下⾯是⼀个例⼦：

--需求：同步MySQL的orders表到TiDB的orders表--1、定义MySQL中orders表的cdc源表CREATETABLEmysql_orders (
idINTNOTNULL,
product_idBIGINT,
...
PRIMARYKEY(id)
) WITH (
'connector'='mysql-cdc',
'hostname'='xx',
'port'='3306',
'username'='xx',
'password'='xx',
'database-name'='xx',
'table-name'='orders');
--2、创建TiDB结果表CREATETABLEtidb_orders(
idINTNOTNULL,
product_idBIGINT,
...
PRIMARYKEY(id)
)
WITH (
'connector'='jdbc',
'url'='jdbc:mysql://localhost:3306/xx',
'table-name'='orders');
--3、从源表读取数据写⼊结果表INSERTINTOtidb_ordersSELECT*FROMmysql_orders

所以基于Flink CDC的⽅案是⼀个纯 SQL 作业，⼤⼤降低了降低了使⽤⻔槛。当然，我们也可以利⽤ Flink SQL 提供的丰富语法进⾏数据清洗、分析、聚合，⽽不仅仅是简单的数据同步。利⽤ Flink SQL 双流 JOIN、维表 JOIN、 UDTF 语法可以⾮常容易地完成实时打宽，以及各种业务逻辑加⼯。

常见CDC方案比较

开源方案	Flink CDC	Debezium	DataX	Canal	Sqoop	Kettle	Oracle Goldengate（OGG）
CDC机制	日志	日志	查询	日志(mysql)	查询	查询	日志(oracle)
增量同步	✓	✓	✓(侵入业务&T+1)	✓	✓(侵入业务&T+1)	✓(侵入业务&T+1)	✓
断点续传	✓	✓	✕	✓	✕	✕	✓
全量同步	✓	✓	✓	✕	✓	✓	✓
全量+增量	✓	✓	✕	✕	✕	✕	✓
架构	分布式	单机	分布式	单机	分布式	分布式	分布式
数据转换/清洗	强	弱	弱	弱	弱	弱	较弱

对⽐增量同步能⼒

基于⽇志的⽅式，可以很好的做到增量同步(准实时)；
⽽基于查询的⽅式必须侵⼊业务才能做到增量同步的，⽽且是T+1的增量同步

对⽐全量同步能⼒，基于查询或者⽇志的 CDC ⽅案基本都⽀持，除了 Canal。

⽽对⽐全量 + 增量同步的能⼒，只有 Flink CDC、Debezium、Oracle Goldengate ⽀持较好。

从架构⻆度去看，该表将架构分为单机和分布式，这⾥的分布式架构不单纯体现在数据读取能⼒的⽔平扩展上，更重要的是在⼤数据场景下分布式系统接⼊能⼒。例如 Flink CDC 的数据⼊湖或者⼊仓的时候，下游通常是分布式的系统，如 Hive、HDFS、Iceberg、Hudi 等，那么从对接⼊分布式系统能⼒上看，Flink CDC 的架构能够很好地接⼊此类系

在数据转换 / 数据清洗能⼒上，当数据进⼊到 CDC ⼯具的时候是否能较⽅便的对数据做⼀些过滤或者清洗，甚⾄聚合在

Flink CDC 上操作相当简单，可以通过 Flink SQL 去操作这些
但是像 DataX、Debezium 等则需要通过脚本或者模板去做，所以⽤户的使⽤⻔槛会⽐较⾼

另外，在⽣态⽅⾯，这⾥指的是下游的⼀些数据库或者数据源的⽀持。Flink CDC 下游有丰富的 Connector，例如写⼊到 TiDB、MySQL、Pg、HBase、Kafka、ClickHouse 等常⻅的⼀些系统，也⽀持各种⾃定义 connector。

使用方式1：DataStream API

引入依赖

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId><version>${flink.version}</version><!--<scope>provided</scope>--></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-planner_${scala.binary.version}</artifactId><version>${flink.version}</version><!--<scope>provided</scope>--></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-scala_${scala.binary.version}</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_${scala.binary.version}</artifactId><version>${flink.version}</version></dependency><!--https://mvnrepository.com/artifact/org.apache.flink/flink-csv --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-csv</artifactId><version>${flink.version}</version></dependency><!--https://mvnrepository.com/artifact/org.apache.flink/flink-json --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-json</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>${hadoop.version}</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.21</version></dependency><dependency><groupId>com.ververica</groupId><artifactId>flink-connector-mysql-cdc</artifactId><version>2.2.0</version></dependency>

样本代码

publicclassFlinkCDCSimple {
publicstaticvoidmain(String[] args) throwsException {
StreamExecutionEnvironmentenv=StreamExecutionEnvironment.getExecutionEnvironment();
//flinkcdc将读取binlog的位置信息以状态的⽅式保存在checkpoint,如果想要做到断点续传,需要从Checkpoint或者Savepoint启动程序//开启Checkpoint（⽣产上是分钟级）env.enableCheckpointing(10000L);
env.getCheckpointConfig().setCheckpointTimeout(20000L);
//指定Checkpoint的⼀致性语义env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//设置任务关闭的时候保留Checkpointenv.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//指定⾃动重启策略env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 2000L));
//设置状态后端env.getCheckpointConfig().setCheckpointStorage("hdfs://node01:8020/flinkCDCbackend");
MySqlSource<String>mySqlSource=MySqlSource.<String>builder()
 .hostname("node01")
 .port(3306)
 .databaseList("cm")
 .tableList("cm.music_style")
 .username("flinkcdc")
 .password("flinkcdc%123")
 .startupOptions(StartupOptions.initial())
 .deserializer(newJsonDebeziumDeserializationSchema()) // convertsSourceRecordtoJSONString// .deserializer(new StringDebeziumDeserializationSchema()) // convertsSourceRecordtoJSONString .build();
env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
// set 4 parallel source tasks .setParallelism(4)
 .print()
env.execute("")

Flink CDC 状态与容错

//flinkcdc将读取binlog的位置信息以状态的⽅式保存在checkpoint,如果想要做到断点续传,需要从Checkpoint或者Savepoint启动程序//开启Checkpointenv.enableCheckpointing(10000L);
env.getCheckpointConfig().setCheckpointTimeout(20000L);
//指定Checkpoint的⼀致性语义env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//设置任务关闭的时候保留Checkpointenv.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//指定⾃动重启策略env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 2000L));
//设置状态后端env.getCheckpointConfig().setCheckpointStorage("hdfs://node01:8020/flinkCDC-backend");

使用方式2：Flink SQL

https://ververica.github.io/flink-cdc-connectors/master/content/connectors/mysql-cdc.html

依赖管理

将flink-sql-connector-mysql-cdc-2.2-SNAPSHOT.jar放到FLINK_HOME/lib下

Flink 全局配置

修改flink-conf.yaml⽂件：

execution.target: yarn-per-job#execution.checkpointing.interval: 3minexecution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATIONexecution.checkpointing.max-concurrent-checkpoints: 1#execution.checkpointing.min-pause: 0execution.checkpointing.mode: EXACTLY_ONCEexecution.checkpointing.timeout: 10min#execution.checkpointing.tolerable-failed-checkpoints: 0#execution.checkpointing.unaligned: false##Supportedbackendsare'jobmanager', 'filesystem', 'rocksdb', orthe#<class-name-of-factory>.
#state.backend: filesystem#Directoryforcheckpointsfilesystem, whenusinganyofthedefaultbundled#statebackends.
#state.checkpoints.dir:hdfs://node01:8020/flinkCDC-checkpoints

sql 代码

SET'execution.checkpointing.interval'='10s';
SET'parallelism.default'='3';
CREATETABLEmusic_style (
music_style_idBIGINT,
style_nameSTRING,
PRIMARYKEY(music_style_id) NOTENFORCED) WITH (
'connector'='mysql-cdc',
'hostname'='node01',
'port'='3306',
'username'='flinkcdc',
'password'='flinkcdc%123',
'database-name'='cm',
'table-name'='music_style',
'connect.timeout'='60s');
CREATETABLEmusic_style_copy (
music_style_idBIGINT,
style_nameSTRING,
PRIMARYKEY(music_style_id) NOTENFORCED) WITH (
'connector'='jdbc',
'url'='jdbc:mysql://node01:3306/test?useUnicode=true&characterEncoding=utf8',
'username'='root',
'password'='root%123',
'table-name'='music_style_copy',
'sink.parallelism'='2');
insertintomusic_style_copySELECT*FROMmusic_style/*+ OPTIONS('server-id'='5401-5404') */ ;

canal job保存save point

flinkstop--savepointPathhdfs://node01:8020/flinkCDC-savepoints -Dyarn.application.id=application_1648361146632_0011e2eebb08b4dac374788f38b290bcf1cf

cacal job之后重新恢复job

SET'execution.checkpointing.interval'='10s';
SET'parallelism.default'='3';
SET'execution.savepoint.path'='hdfs://node01:8020/flinkCDC-savepoints/savepointe2eebb-3065002c8658';
CREATETABLEmusic_style (
music_style_idBIGINT,
style_nameSTRING,
PRIMARYKEY(music_style_id) NOTENFORCED) WITH (
'connector'='mysql-cdc',
'hostname'='node01',
'port'='3306',
'username'='flinkcdc',
'password'='flinkcdc%123',
'database-name'='cm',
'table-name'='music_style',
'connect.timeout'='60s');
CREATETABLEmusic_style_copy (
music_style_idBIGINT,
style_nameSTRING,
PRIMARYKEY(music_style_id) NOTENFORCED) WITH (
'connector'='jdbc',
'url'='jdbc:mysql://node01:3306/test',
'username'='root',
'password'='root%123',
'table-name'='music_style_copy');
insertintomusic_style_copySELECT*FROMmusic_style/*+ OPTIONS('server-id'='5401-5404') */ ;

Flink CDC 2.xx设计

参考：https://blog.csdn.net/qq_30438573/article/details/119078255

Flink CDC使用(数据采集CDC方案比较)

什么是CDC

cdc使用场景

cdc 实现

Flink CDC

⽀持的连接器

支持Flink 版本

Flink CDC 优势

Flink cdc采集方案

常见CDC方案比较

使用方式1：DataStream API

Flink CDC 状态与容错

使用方式2：Flink SQL

依赖管理

Flink 全局配置

sql 代码

canal job保存save point

cacal job之后重新恢复job

Flink CDC 2.xx设计

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

Flink CDC使用(数据采集CDC方案比较)

什么是CDC

cdc使用场景

cdc 实现

Flink CDC

⽀持的连接器

支持Flink 版本

Flink CDC 优势

Flink cdc采集方案

常见CDC方案比较

使用方式1：DataStream API

Flink CDC 状态与容错

使用方式2：Flink SQL

依赖管理

Flink 全局配置

sql 代码

canal job保存save point

cacal job之后重新恢复job

Flink CDC 2.xx设计

热门文章

最新文章

相关课程

相关电子书

相关实验场景