devbox
Monodyee
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Flink CDC介绍_flinkcdc

作者：Monodyee | 2024-04-07 22:35:23

赞

踩

flinkcdc

1.CDC概述


CDC（Change Data Capture）是一种用于捕获和处理数据源中的变化的技术。它允许实时地监视数据库或数据流中发生的数据变动，并将这些变动抽取出来，以便进行进一步的处理和分析。

传统上，数据源的变化通常通过周期性地轮询整个数据集进行检查来实现。但是，这种轮询的方式效率低下且不能实时反应变化。而 CDC 技术则通过在数据源上设置一种机制，使得变化的数据可以被实时捕获并传递给下游处理系统，从而实现了实时的数据变动监控。

Flink 作为一个强大的流式计算引擎，提供了内置的 CDC 功能，能够连接到各种数据源（如数据库、消息队列等），捕获其中的数据变化，并进行灵活的实时处理和分析。

通过使用 Flink CDC，我们可以轻松地构建实时数据管道，对数据变动进行实时响应和处理，为实时分析、实时报表和实时决策等场景提供强大的支持。

Flink_CDC

 

2.CDC 的实现原理


通常来讲，CDC 分为主动查询和事件接收两种技术实现模式。对于主动查询而言，用户通常会在数据源表的某个字段中，保存上次更新的时间戳或版本号等信息，然后下游通过不断的查询和与上次的记录做对比，来确定数据是否有变动，是否需要同步。这种方式优点是不涉及数据库底层特性，实现比较通用；缺点是要对业务表做改造，且实时性不高，不能确保跟踪到所有的变更记录，且持续的频繁查询对数据库的压力较大。事件接收模式可以通过触发器（Trigger）或者日志（例如 Transaction log、Binary log、Write-ahead log 等）来实现。当数据源表发生变动时，会通过附加在表上的触发器或者 binlog 等途径，将操作记录下来。下游可以通过数据库底层的协议，订阅并消费这些事件，然后对数据库变动记录做重放，从而实现同步。这种方式的优点是实时性高，可以精确捕捉上游的各种变动；缺点是部署数据库的事件接收和解析器（例如 Debezium、Canal 等），有一定的学习和运维成本，对一些冷门的数据库支持不够。综合来看，事件接收模式整体在实时性、吞吐量方面占优，如果数据源是 MySQL、PostgreSQL、MongoDB 等常见的数据库实现，建议使用Debezium来实现变更数据的捕获（下图来自Debezium 官方文档如果使用的只有 MySQL，则还可以用Canal。
在这里插入图片描述

 

3.为什么选 Flink
从上图可以看到，Debezium 官方架构图中，是通过 Kafka Streams 直接实现的 CDC 功能。而我们这里更建议使用 Flink CDC 模块，因为 Flink 相对 Kafka Streams 而言，有如下优势：

强大的流处理引擎： Flink 是一个强大的流处理引擎，具备高吞吐量、低延迟、Exactly-Once 语义等特性。它通过基于事件时间的处理模型，支持准确和有序的数据处理，适用于实时数据处理和分析场景。这使得 Flink 成为实现 CDC 的理想选择。

内置的 CDC 功能： Flink 提供了内置的 CDC 功能，可以直接连接到各种数据源，捕获数据变化，并将其作为数据流进行处理。这消除了我们自行开发或集成 CDC 解决方案的需要，使得实现 CDC 变得更加简单和高效。

多种数据源的支持： Flink CDC 支持与各种数据源进行集成，如关系型数据库（如MySQL、PostgreSQL）、消息队列（如Kafka、RabbitMQ）、文件系统等。这意味着无论你的数据存储在哪里，Flink 都能够轻松地捕获其中的数据变化，并进行进一步的实时处理和分析。

灵活的数据处理能力： Flink 提供了灵活且强大的数据处理能力，可以通过编写自定义的转换函数、处理函数等来对 CDC 数据进行各种实时计算和分析。同时，Flink 还集成了 SQL 和 Table API，为用户提供了使用 SQL 查询语句或 Table API 进行简单查询和分析的方式。

完善的生态系统： Flink 拥有活跃的社区和庞大的生态系统，这意味着你可以轻松地获取到丰富的文档、教程、示例代码和解决方案。此外，Flink 还与其他流行的开源项目（如Apache Kafka、Elasticsearch）深度集成，提供了更多的功能和灵活性。

4.支持的连接器

 5.支持的 Flink 版本

 

6.Flink CDC特性
支持读取数据库快照，即使出现故障也能继续读取binlog，并进行Exactly-once处理
DataStream API 的 CDC 连接器，用户可以在单个作业中使用多个数据库和表的更改，而无需部署 Debezium 和 Kafka
Table/SQL API 的 CDC 连接器，用户可以使用 SQL DDL 创建 CDC 源来监视单个表上的更改
下表显示了连接器的当前特性：

 

7.用法实例

7.1DataStream API 的用法(推荐)

请严格按照上面的《5.支持的 Flink 版本》搭配来使用Flink CDC

  1. <properties>
  2. 	<flink.version>1.13.0</flink.version>
  3. 	<maven.compiler.source>1.8</maven.compiler.source>
  4. 	<maven.compiler.target>1.8</maven.compiler.target>
  5. </properties>
  6. <dependency>
  7. 	<groupId>com.ververica</groupId>
  8. 	<artifactId>flink-connector-mysql-cdc</artifactId>
  9. 	<version>${flinkcdc.version}</version>
  10. </dependency>
  11. <!-- flink核心API -->
  12. <dependency>
  13. 	<groupId>org.apache.flink</groupId>
  14. 	<artifactId>flink-clients_2.12</artifactId>
  15. 	<version>${flink.version}</version>
  16. </dependency>
  17. <dependency>
  18. 	<groupId>org.apache.flink</groupId>
  19. 	<artifactId>flink-java</artifactId>
  20. 	<version>${flink.version}</version>
  21. </dependency>
  22. <dependency>
  23. 	<groupId>org.apache.flink</groupId>
  24. 	<artifactId>flink-scala_2.12</artifactId>
  25. 	<version>${flink.version}</version>
  26. </dependency>
  27. <dependency>
  28. 	<groupId>org.apache.flink</groupId>
  29. 	<artifactId>flink-streaming-java_2.12</artifactId>
  30. 	<version>${flink.version}</version>
  31. </dependency>
  32. <dependency>
  33. 	<groupId>org.apache.flink</groupId>
  34. 	<artifactId>flink-streaming-scala_2.12</artifactId>
  35. 	<version>${flink.version}</version>
  36. </dependency>
  37. <dependency>
  38. 	<groupId>org.apache.flink</groupId>
  39. 	<artifactId>flink-table-common</artifactId>
  40. 	<version>${flink.version}</version>
  41. </dependency>
  42. <dependency>
  43. 	<groupId>org.apache.flink</groupId>
  44. 	<artifactId>flink-table-planner-blink_2.12</artifactId>
  45. 	<version>${flink.version}</version>
  46. </dependency>
  47. <dependency>
  48. 	<groupId>org.apache.flink</groupId>
  49. 	<artifactId>flink-table-api-java-bridge_2.12</artifactId>
  50. 	<version>${flink.version}</version>
  51. </dependency>

请提前开启MySQL中的binlog，配置my.cnf文件，重启mysqld服务即可

my.cnf

  1. [client]
  2. default_character_set=utf8
  3. [mysqld]
  4. server-id=1
  5. collation_server=utf8_general_ci
  6. character_set_server=utf8
  7. log-bin=mysql-bin
  8. binlog_format=row
  9. expire_logs_days=30

ddl&dml.sql

  1. create table test_cdc
  2. (
  3.     id   int          not null
  4.         primary key,
  5.     name varchar(100) null,
  6.     age  int          null
  7. );
  8.  
  9. INSERT INTO flink.test_cdc (id, name, age) VALUES (1, 'Daniel', 25);
  10. INSERT INTO flink.test_cdc (id, name, age) VALUES (2, 'David', 38);
  11. INSERT INTO flink.test_cdc (id, name, age) VALUES (3, 'James', 16);
  12. INSERT INTO flink.test_cdc (id, name, age) VALUES (4, 'Robert', 27);

FlinkDSCDC.java

  1. package com.daniel.util;
  2.  
  3. import com.ververica.cdc.connectors.mysql.MySqlSource;
  4. import com.ververica.cdc.connectors.mysql.table.StartupOptions;
  5. import com.ververica.cdc.debezium.DebeziumSourceFunction;
  6. import com.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
  7. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  9.  
  10. /**
  11.  * @Author Daniel
  12.  * @Date: 2023/7/25 10:03
  13.  * @Description DataStream API CDC
  14.  **/
  15. public class FlinkDSCDC {
  16.     public static void main(String[] args) throws Exception {
  17.  
  18.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  19.         env.setParallelism(1);
  20.  
  21.         DebeziumSourceFunction<String> sourceFunction = MySqlSource.<String>builder()
  22.                 .hostname("localhost")
  23.                 .port(3306)
  24.                 .username("root")
  25.                 .password("123456")
  26.                 .databaseList("flink")
  27.                 // 这里一定要是db.table的形式
  28.                 .tableList("flink.test_cdc")
  29.                 .deserializer(new StringDebeziumDeserializationSchema())
  30.                 .startupOptions(StartupOptions.initial())
  31.                 .build();
  32.  
  33.         DataStreamSource<String> dataStreamSource = env.addSource(sourceFunction);
  34.         dataStreamSource.print();
  35.         env.execute("FlinkDSCDC");
  36.     }
  37. }
  38.  
  1. UPDATE flink.test_cdc t SET t.age = 24 WHERE t.id = 1;
  2. UPDATE flink.test_cdc t SET t.name = 'Andy' WHERE t.id = 3;

打印出的日志

  1. SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={transaction_id=null, ts_sec=1690272544, file=mysql-bin.000001, pos=7860, row=1, server_id=1, event=2}} ConnectRecord{topic='mysql_binlog_source.flink.test_cdc', kafkaPartition=null, key=Struct{id=1}, keySchema=Schema{mysql_binlog_source.flink.test_cdc.Key:STRUCT}, value=Struct{before=Struct{id=1,name=Daniel,age=25},after=Struct{id=1,name=Daniel,age=24},source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1690272544000,db=flink,table=test_cdc,server_id=1,file=mysql-bin.000001,pos=7989,row=0},op=u,ts_ms=1690272544122}, valueSchema=Schema{mysql_binlog_source.flink.test_cdc.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
  2. SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={transaction_id=null, ts_sec=1690272544, file=mysql-bin.000001, pos=7860, row=1, server_id=1, event=4}} ConnectRecord{topic='mysql_binlog_source.flink.test_cdc', kafkaPartition=null, key=Struct{id=3}, keySchema=Schema{mysql_binlog_source.flink.test_cdc.Key:STRUCT}, value=Struct{before=Struct{id=3,name=James,age=16},after=Struct{id=3,name=Andy,age=16},source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1690272544000,db=flink,table=test_cdc,server_id=1,file=mysql-bin.000001,pos=8113,row=0},op=u,ts_ms=1690272544122}, valueSchema=Schema{mysql_binlog_source.flink.test_cdc.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}

可以得出的结论：

1、日志中的数据变化操作类型（op）可以表示为 ‘u’，表示更新操作。在第一条日志中，发生了一个更新操作，对应的记录的 key 是 id=1，更新前的数据是 {id=1, name=Daniel, age=25}，更新后的数据是 {id=1, name=Daniel, age=24}。在第二条日志中，也发生了一个更新操作，对应的记录的 key 是 id=3，更新前的数据是 {id=3, name=James, age=16}，更新后的数据是 {id=3, name=Andy, age=16}。
2、每条日志还提供了其他元数据信息，如数据源（source）、版本号（version）、连接器名称（connector）、时间戳（ts_ms）等。这些信息可以帮助我们追踪记录的来源和处理过程。
3、日志中的 sourceOffset 包含了一些关键信息，如事务ID（transaction_id）、文件名（file）、偏移位置（pos）等。这些信息可以用于确保数据的准确顺序和一致性。

7.2Table/SQL API的用法

FlinkSQLCDC.java

  1. package com.daniel.util;
  2.  
  3. import org.apache.flink.api.java.tuple.Tuple2;
  4. import org.apache.flink.streaming.api.datastream.DataStream;
  5. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  6. import org.apache.flink.table.api.Table;
  7. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  8. import org.apache.flink.types.Row;
  9.  
  10. /**
  11.  * @Author Daniel
  12.  * @Date: 2023/7/25 15:25
  13.  * @Description
  14.  **/
  15. public class FlinkSQLCDC {
  16.     public static void main(String[] args) throws Exception {
  17.  
  18.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  19.         env.setParallelism(1);
  20.  
  21.         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
  22.         tableEnv.executeSql("CREATE TABLE test_cdc (" +
  23.                 " id int primary key," +
  24.                 " name STRING," +
  25.                 " age int" +
  26.                 ") WITH (" +
  27.                 " 'connector' = 'mysql-cdc'," +
  28.                 " 'scan.startup.mode' = 'latest-offset'," +
  29.                 " 'hostname' = 'localhost'," +
  30.                 " 'port' = '3306'," +
  31.                 " 'username' = 'root'," +
  32.                 " 'password' = '123456'," +
  33.                 " 'database-name' = 'flink'," +
  34.                 " 'table-name' = 'test_cdc'" +
  35.                 ")");
  36.  
  37.         Table table = tableEnv.sqlQuery("select * from test_cdc");
  38.         DataStream<Tuple2<Boolean, Row>> dataStreamSource = tableEnv.toRetractStream(table, Row.class);
  39.         dataStreamSource.print();
  40.         env.execute("FlinkSQLCDC");
  41.     }
  42. }
  43.  
  1. UPDATE flink.test_cdc t SET t.age = 55 WHERE t.id = 2;
  2. UPDATE flink.test_cdc t SET t.age = 22 WHERE t.id = 3;
  3. UPDATE flink.test_cdc t SET t.name = 'Alice' WHERE t.id = 4;
  4. UPDATE flink.test_cdc t SET t.age = 18 WHERE t.id = 1;
  5. INSERT INTO flink.test_cdc (id, name, age) VALUES (5, 'David', 29);

打印出的日志

  1. (false,-U[2, David, 38])
  2. (true,+U[2, David, 55])
  3. (false,-U[3, Andy, 16])
  4. (true,+U[3, Andy, 22])
  5. (false,-U[4, Robert, 27])
  6. (true,+U[4, Alice, 27])
  7. (false,-U[1, Daniel, 24])
  8. (true,+U[1, Daniel, 18])
  9. (true,+I[5, David, 29])

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Monodyee/article/detail/381084
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号