经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 大数据/云/AI » Flink » 查看文章
一文解开主流开源变更数据捕获技术之Flink CDC的入门使用
来源：cnblogs　　作者：itxiaoshen　　时间：2023/6/28 8:50:01　　对本文有异议

@

目录

概述

定义

flink-cdc-connectors 官网 https://github.com/ververica/flink-cdc-connectors 源码release最新版本2.4.0

flink-cdc-connectors 文档地址 https://ververica.github.io/flink-cdc-connectors/master/

flink-cdc-connectors 源码地址 https://github.com/ververica/flink-cdc-connectors

CDC Connectors for Apache Flink 是Apache Flink的一组源连接器，使用更改数据捕获(CDC)从不同的数据库摄取更改，其集成了Debezium作为捕获数据变化的引擎，因此它可以充分利用Debezium的能力。

Flink CDC是由Flink社区开发的flink-cdc-connectors 的source组件，基于数据库日志的 Change Data Caputre 技术，实现了从 MySQL、PostgreSQL 等数据库全量和增量的一体化读取能力，并借助 Flink 优秀的管道能力和丰富的上下游生态，支持捕获多种数据库的变更，并将这些变更实时同步到下游存储。

什么是CDC?

这里也简单说明下，CDC为三个英文Change Data Capture（变更数据捕获）的缩写，核心思想是监测并捕获数据库的变动（包括数据或数据表的插入、更新以及删除等），将这些变更按发生的顺序完整记录下来，写入到消息中间件中以供其它服务进行订阅及消费。

CDC的分类

CDC主要分为基于查询的CDC和基于binlog的CDC，两者之间区别主要如下：

image-20230626155037373

特性

  • 支持读取数据库快照，即使发生故障，也只进行一次处理，继续读取事务日志。
  • 数据流API的CDC连接器，用户可以在单个作业中消费多个数据库和表上的更改，而无需部署Debezium和Kafka。
  • 用于表/SQL API的CDC连接器，用户可以使用SQL DDL创建CDC源来监视单个表上的更改。

应用场景

  • 数据分发，将一个数据源分发给多个下游，常用于业务解耦、微服务。
  • 数据集成，将分散异构的数据源集成到数据仓库中，消除数据孤岛，便于后续的分析。
  • 数据迁移，常用于数据库备份、容灾等。

支持数据源

CDC Connectors for Apache Flink支持从多种数据库到Flink摄取快照数据和实时更改，然后转换和下沉到各种下游系统

image-20230626154109523

支撑数据源包括如下：

image-20230626160745676

实战

这里以MySQL作为数据源为例，通过flink-connector-mysql-cdc实现数据变更获取，先准备MySQL环境，这里复用前面<<实时采集MySQL数据之轻量工具Maxwell实操>>的文章环境，数据库有两个my_maxwell_01，my_maxwell_02，每个数据库都有相同account和product表。pom文件引入依赖

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <modelVersion>4.0.0</modelVersion>
  7.     <groupId>cn.itxs.flink</groupId>
  8.     <artifactId>flink-cdc-demo</artifactId>
  9.     <version>1.0-SNAPSHOT</version>
  11.     <properties>
  12.         <maven.compiler.source>8</maven.compiler.source>
  13.         <maven.compiler.target>8</maven.compiler.target>
  14.         <flink.version>1.17.1</flink.version>
  15.         <flink.cdc.version>2.4.0</flink.cdc.version>
  16.         <mysql.client.version>8.0.29</mysql.client.version>
  17.         <fastjson.version>1.2.83</fastjson.version>
  18.     </properties>
  20.     <dependencies>
  21.         <dependency>
  22.             <groupId>org.apache.flink</groupId>
  23.             <artifactId>flink-streaming-java</artifactId>
  24.             <version>${flink.version}</version>
  25.             <scope>provided</scope>
  26.         </dependency>
  27.         <dependency>
  28.             <groupId>org.apache.flink</groupId>
  29.             <artifactId>flink-clients</artifactId>
  30.             <version>${flink.version}</version>
  31.             <scope>provided</scope>
  32.         </dependency>
  33.         <dependency>
  34.             <groupId>org.apache.flink</groupId>
  35.             <artifactId>flink-connector-base</artifactId>
  36.             <version>${flink.version}</version>
  37.             <scope>provided</scope>
  38.         </dependency>
  39.         <dependency>
  40.             <groupId>org.apache.flink</groupId>
  41.             <artifactId>flink-table-api-java-bridge</artifactId>
  42.             <version>${flink.version}</version>
  43.             <scope>provided</scope>
  44.         </dependency>
  45.         <dependency>
  46.             <groupId>org.apache.flink</groupId>
  47.             <artifactId>flink-table-planner-loader</artifactId>
  48.             <version>${flink.version}</version>
  49.             <scope>provided</scope>
  50.         </dependency>
  51.         <dependency>
  52.             <groupId>org.apache.flink</groupId>
  53.             <artifactId>flink-table-runtime</artifactId>
  54.             <version>${flink.version}</version>
  55.             <scope>provided</scope>
  56.         </dependency>
  57.         <dependency>
  58.             <groupId>mysql</groupId>
  59.             <artifactId>mysql-connector-java</artifactId>
  60.             <version>${mysql.client.version}</version>
  61.         </dependency>
  62.         <dependency>
  63.             <groupId>com.ververica</groupId>
  64.             <artifactId>flink-connector-mysql-cdc</artifactId>
  65.             <version>${flink.cdc.version}</version>
  66.         </dependency>
  67.         <dependency>
  68.             <groupId>com.alibaba</groupId>
  69.             <artifactId>fastjson</artifactId>
  70.             <version>${fastjson.version}</version>
  71.         </dependency>
  72.     </dependencies>
  74.     <build>
  75.         <plugins>
  76.             <plugin>
  77.                 <groupId>org.apache.maven.plugins</groupId>
  78.                 <artifactId>maven-shade-plugin</artifactId>
  79.                 <version>3.2.4</version>
  80.                 <executions>
  81.                     <execution>
  82.                         <phase>package</phase>
  83.                         <goals>
  84.                             <goal>shade</goal>
  85.                         </goals>
  86.                         <configuration>
  87.                             <artifactSet>
  88.                                 <excludes>
  89.                                     <exclude>com.google.code.findbugs:jsr305</exclude>
  90.                                     <exclude>org.slf4j:*</exclude>
  91.                                     <exclude>log4j:*</exclude>
  92.                                 </excludes>
  93.                             </artifactSet>
  94.                             <filters>
  95.                                 <filter>
  96.                                     <!-- Do not copy the signatures in the META-INF folder.
  97.                                     Otherwise, this might cause SecurityExceptions when using the JAR. -->
  98.                                     <artifact>*:*</artifact>
  99.                                     <excludes>
  100.                                         <exclude>META-INF/*.SF</exclude>
  101.                                         <exclude>META-INF/*.DSA</exclude>
  102.                                         <exclude>META-INF/*.RSA</exclude>
  103.                                     </excludes>
  104.                                 </filter>
  105.                             </filters>
  106.                             <transformers combine.children="append">
  107.                                 <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer">
  108.                                 </transformer>
  109.                             </transformers>
  110.                         </configuration>
  111.                     </execution>
  112.                 </executions>
  113.             </plugin>
  114.         </plugins>
  115.     </build>
  116. </project>

创建DataStreamDemo.java，

  1. package cn.itxs.cdc;
  3. import com.ververica.cdc.connectors.mysql.source.MySqlSource;
  4. import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
  5. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  8. public class DataStreamDemo {
  9.     public static void main(String[] args) throws Exception {
  10.         MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
  11.                 .hostname("mysqlserver")
  12.                 .port(3306)
  13.                 .databaseList("my_maxwell_01,my_maxwell_02")
  14.                 .tableList("my_maxwell_01.*,my_maxwell_02.product")
  15.                 .username("root")
  16.                 .password("12345678")
  17.                 .deserializer(new JsonDebeziumDeserializationSchema()) // 将SourceRecord转换为JSON字符串
  18.                 .build();
  20.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  22.         // 开启checkpoint
  23.         env.enableCheckpointing(3000);
  25.         env
  26.                 .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
  27.                 // 设置平行度为4
  28.                 .setParallelism(4)
  29.                 .print().setParallelism(1); // 对sink打印使用并行性1来保持消息顺序
  31.         env.execute("Print MySQL Snapshot + Binlog");
  32.     }
  33. }

由于上面flink的依赖配置provided，因此在IDEA中启动的话需要勾选下面标红的选项

image-20230627165925042

启动程序，查看日志可以看到从mysql读取目前全量的数据，my_maxwell_02也只读取product表数据

image-20230627170148732

修改两个库的表后可以看到相应修改信息，其中也确认my_maxwell_02的account没有读取变更数据。

image-20230627170435984

  1. {"before":{"id":7,"name":"李丹","age":44},"after":{"id":7,"name":"李丹","age":48},"source":{"version":"1.9.7.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1687856595000,"snapshot":"false","db":"my_maxwell_01","sequence":null,"table":"account","server_id":1,"gtid":null,"file":"binlog.000025","pos":2798,"row":0,"thread":330184,"query":null},"op":"u","ts_ms":1687856598620,"transaction":null}
  2. {"before":{"id":1,"name":"iphone13","type":1},"after":{"id":1,"name":"iphone14","type":1},"source":{"version":"1.9.7.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1687856605000,"snapshot":"false","db":"my_maxwell_01","sequence":null,"table":"product","server_id":1,"gtid":null,"file":"binlog.000025","pos":3140,"row":0,"thread":330184,"query":null},"op":"u","ts_ms":1687856608748,"transaction":null}
  3. {"before":{"id":1,"name":"iphone13","type":1},"after":{"id":1,"name":"iphone14","type":1},"source":{"version":"1.9.7.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1687856628000,"snapshot":"false","db":"my_maxwell_02","sequence":null,"table":"product","server_id":1,"gtid":null,"file":"binlog.000025","pos":3486,"row":0,"thread":330184,"query":null},"op":"u","ts_ms":1687856631643,"transaction":null}

打包后放到集群上，执行

  1. bin/flink run -m hadoop1:8081 -c cn.itxs.cdc.DataStreamDemo ./lib/flink-cdc-demo-1.0-SNAPSHOT.jar

image-20230627185543185

可以看到的日志也成功输出表的全量的日志和刚才修改增量数据

image-20230627185508643

如果需要断点续传可以使用状态后端存储来实现

  1.         CheckpointConfig checkpointConfig = env.getCheckpointConfig();
  2.         checkpointConfig.setCheckpointStorage("hdfs://hadoop111:9000/checkpoints/flink/cdc");
  3.         checkpointConfig.setMinPauseBetweenCheckpoints(TimeUnit.SECONDS.toMillis(2));
  4.         checkpointConfig.setTolerableCheckpointFailureNumber(5);
  5.         checkpointConfig.setCheckpointTimeout(TimeUnit.MINUTES.toMillis(1));
  6.         checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

FlinkSQL方式代码示例

创建SqlDemo.java文件

  1. package cn.itxs.cdc;
  3. import org.apache.flink.streaming.api.datastream.DataStream;
  4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  5. import org.apache.flink.table.api.Table;
  6. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  7. import org.apache.flink.types.Row;
  9. public class SqlDemo {
  10.     public static void main(String[] args) throws Exception {
  11.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  12.         env.setParallelism(1);
  13.         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
  14.         tableEnv.executeSql("CREATE TABLE account (\n" +
  15.                 " id INT NOT NULL,\n" +
  16.                 " name STRING,\n" +
  17.                 " age INT,\n" +
  18.                 " PRIMARY KEY(id) NOT ENFORCED\n" +
  19.                 ") WITH (\n" +
  20.                 " 'connector' = 'mysql-cdc',\n" +
  21.                 " 'hostname' = 'mysqlserver',\n" +
  22.                 " 'port' = '3306',\n" +
  23.                 " 'username' = 'root',\n" +
  24.                 " 'password' = '12345678',\n" +
  25.                 " 'database-name' = 'my_maxwell_01',\n" +
  26.                 " 'table-name' = 'account'\n" +
  27.                 ");");
  29.         Table table = tableEnv.sqlQuery("select * from account");
  30.         DataStream<Row> rowDataStream = tableEnv.toChangelogStream(table);
  31.         rowDataStream.print("account_binlog====");
  32.         env.execute();
  33.     }
  34. }

启动程序，查看日志可以看到从mysql读取my_maxwell_01库account表的全量的数据，修改表数据也确认读取变更数据。

image-20230627182647771

  • 本人博客网站IT小神 www.itxiaoshen.com

原文链接：https://www.cnblogs.com/itxiaoshen/p/17510226.html

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Flink热门文章

  • 深度介绍Flink在字节跳动数据流的实践
  • idea中flink启动报错org.apache.flink.api.common.ExecutionConfig$GlobalJobParameters
  • flink-cdc同步mysql数据到elasticsearch
  • 实时计算框架：Flink集群搭建与运行机制
  • Flink的Job启动Driver端(源码分析)
  • Flink中TaskManager端执行用户逻辑过程(源码分析)
  • 统一批处理流处理——Flink批流一体实现原理
  • Flink+Druid构建实时OLAP的探索
  • 01.Flink笔记-编译、部署
  • 一文解开主流开源变更数据捕获技术之Flink CDC的入门使用

    • Flink推荐文章

  • Flink Batch Hash Aggregate
  • flink-cdc同步mysql数据到elasticsearch
  • 一文解开主流开源变更数据捕获技术之Flink CDC的入门使用
  • 实例讲解Flink 流处理程序编程模型
  • 《关于我因为flink成为spark源码贡献者这件小事》
  • Flink SQL管理平台flink-streaming-platform-web安装搭建
  • 深度介绍Flink在字节跳动数据流的实践
  • 实时计算框架：Flink集群搭建与运行机制
  • Flink-3-ApacheFlink架构
  • Flink-5-DataStreamAPI

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号