什么是Flink CDC，以及怎样利用

CDC介绍

数据库中的CDC（Change Data Capture，变动数据捕捉）是一种用于实时跟踪数据库中数据变革的技术。CDC的主要目标是在数据库中捕捉增量数据，以便在必要时可以轻松地将这些数据归并到其他系统或应用程序中。CDC在数据库管理、数据同步、数据集成和数据备份等方面具有广泛的应用。
CDC通常通过以下几种方式实现：

• 基于触发器的CDC：在表上创建触发器，当数据发生更改时，触发器会将更改的数据记录到其他系统或表中。
  
• 基于事件日志的CDC：通过读取数据库事件日志，将日志中的更改记录剖析为可操作的数据。这种方法通常用于增量备份和恢复。
  
• 基于游标的CDC：在数据库中利用游标，逐行处理数据更改，并将这些更改应用于其他系统或表。
  
• 基于时间戳的CDC：为表中的每个数据行分配一个时间戳，当数据发生更改时，更新相应的时间戳。然后，可以利用时间戳来识别和处理数据更改。
  
• 基于消息队列的CDC：将数据更改作为事件发送到消息队列，以便其他系统或应用程序可以订阅和处理这些事件。
  

Flink CDC

Flink CDC（Change Data Capture，即数据变动抓取）是一个开源的数据库变动日志捕捉和处理框架，它可以实时地从各种数据库（如MySQL、PostgreSQL、Oracle、MongoDB等）中捕捉数据变动并将其转换为流式数据。Flink CDC 可以资助实时应用程序实时地处理和分析这些流数据，从而实现数据同步、数据管道、实时分析和实时应用等功能。
Flink CDC 的主要特点包括：

• 支持多种数据库范例：Flink CDC 支持多种数据库，如 MySQL、PostgreSQL、Oracle、MongoDB 等。
  
• 实时数据捕捉：Flink CDC 可以或许实时捕捉数据库中的数据变动，并将其转换为流式数据。
  
• 高性能：Flink CDC 基于 Flink 引擎，具有高性能的数据处理能力。
  
• 低延迟：Flink CDC 可以在毫秒级的延迟下处理大量的数据变动。
  
• 易集成：Flink CDC 与 Flink 生态系统精密集成，可以方便地与其他 Flink 应用程序一起利用。
  
• 高可用性：Flink CDC 支持实时备份和恢复，确保数据的高可用性。
  

适用于场景？

Flink CDC 可以用于各种场景，如：

• 实时数据同步：将数据从一个数据库实时同步到另一个数据库。
  
• 实时数据管道：构建实时数据处理管道，处理和分析数据库中的数据。
  
• 实时数据分析：实时分析数据库中的数据，提供实时的业务洞察。
  
• 实时应用：将数据库中的数据实时应用于实时应用程序，如实时报表、实时保举等。
  
• 实时监控：实时监控数据库中的数据，检测异常和错误。
  

Flink CDC 的简单用例

数据库设置

创建数据库和相应的表

创建mydb数据库，并创建user表

1. create database mydb;
2. create table user(
3.     id bigint primary key auto_increment,
4.     name varchar(255)
5. );
6. INSERT INTO mydb.user (name) VALUES ('小明');
7. INSERT INTO mydb.user (name) VALUES ('小红');

复制代码

创建了一个名为 mydb 的数据库，并在此中创建了一个名为 user 的表。表中包罗一个主键 id 和一个字符串范例的 name 字段。还向 user 表中插入了两条记录，分别是 '小明' 和 '小红'。
开启mysql数据库bin-log日志

1.假如是服务器

在my.cnf中添加binlog设置,并重启mysql数据库

3.     server-id = 123
4.     log_bin = mysql-bin
5.     binlog_format = row
6.     binlog_row_image = full
7.     expire_logs_days = 10
8.     gtid_mode = on
9.     enforce_gtid_consistency = on
10.    

复制代码

已经为 MySQL 设置了一些设置参数。下面是对这些参数的解释：

• server-id = 123：指定服务器的唯一标识符，通常用于区分不同的数据库服务器。
  
• log_bin = mysql-bin：启用二进制日志记录，以便在数据库出现故障时可以恢复数据。
  
• binlog_format = row：指定二进制日志的记录格式。row 格式会记录每个更改行的详细信息，这对于必要事件完整性的应用程序非常有用。
  
• binlog_row_image = full：设置 row 格式的二进制日志记录行的完整信息，包括列值、注释等。这有助于进步应用程序的可恢复性。
  
• expire_logs_days = 10：设置自动清算逾期二进制日志文件的天数。在这个例子中，设置为 10 天。
  
• gtid_mode = on：启用全局事件 ID 模式，这使得基于 GTID 的复制成为大概。
  
• enforce_gtid_consistency = on：逼迫执行 GTID 一致性，确保事件在不同的 MySQL 实例之间保持一致。
  

2.假如在Windows利用小皮

在小皮面板里设置，如图：

打开bin日志开关
搭建Flink CDC java环境

添加maven相干pom

在pom里添加相干Flink CDC依赖

1. <!--        flink connector 基础包-->
2.         <dependency>
3.             <groupId>org.apache.flink</groupId>
4.             <artifactId>flink-connector-base</artifactId>
5.             <version>1.14.4</version>
6.         </dependency>
7. <!--        CDC mysql 源-->
8.         <dependency>
9.             <groupId>com.ververica</groupId>
10.             <artifactId>flink-sql-connector-mysql-cdc</artifactId>
11.             <version>2.3.0</version>
12.         </dependency>
13. <!--        Flink Steam流处理-->
14.         <dependency>
15.             <groupId>org.apache.flink</groupId>
16.             <artifactId>flink-streaming-java_2.11</artifactId>
17.             <version>1.14.4</version>
18.         </dependency>
19. <!--        flink java客户端-->
20.         <dependency>
21.             <groupId>org.apache.flink</groupId>
22.             <artifactId>flink-clients_2.12</artifactId>
23.             <version>1.14.4</version>
24.         </dependency>
25. <!--        开启webui支持，默认是8081，默认没有开启-->
26.         <dependency>
27.             <groupId>org.apache.flink</groupId>
28.             <artifactId>flink-runtime-web_2.12</artifactId>
29.             <version>1.14.4</version>
30.         </dependency>
31. <!--        Flink Table API和SQL API使得在Flink中进行数据处理变得更加简单和高效
32. 通过使用Table API和SQL API，可以像使用传统的关系型数据库一样，通过编写SQL语句或者使用类似于
33. Java的API进行数据处理和分析-->
34.         <dependency>
35.             <groupId>org.apache.flink</groupId>
36.             <artifactId>flink-table-runtime_2.11</artifactId>
37.             <version>1.14.4</version>
38.         </dependency>
39.         <dependency>
40.             <groupId>ch.qos.logback</groupId>
41.             <artifactId>logback-classic</artifactId>
42.             <version>1.2.11</version>
43.         </dependency>
44.         <dependency>
45.             <groupId>org.slf4j</groupId>
46.             <artifactId>slf4j-api</artifactId>
47.             <version>2.0.6</version>
48.         </dependency>

复制代码

这是一段 Maven 依赖设置，它引入了 Flink Connector Base、CDC MySQL Source、Flink Streaming Java、Flink Java Client、Flink Runtime Web、Flink Table Runtime 和 Logback Classic。
这些依赖库提供了以下功能：

• Flink Connector Base:Flink 的连接器基础包，用于将 Flink 与其他系统举行集成。
  
• CDC MySQL Source:Flink 的 MySQL CDC 源，用于从 MySQL 数据库中读取数据流。
  
• Flink Streaming Java:Flink 的 Java 流处理 API,用于编写并发程序以处理数据流。
  
• Flink Java Client:Flink 的 Java API,用于在 Java 应用程序中利用 Flink。
  
• Flink Runtime Web:Flink 的 Web UI,用于监控和管理 Flink 集群。
  
• Flink Table Runtime:Flink 的 Table API,使在 Flink 中举行数据处理变得更加简单和高效。
  
• Logback Classic:日志记录库，用于记录应用程序的日志信息。
  

构建Sink

Flink CDC(Change Data Capture)中的Sink用于将CDC吸收到的数据写入外部系统(如数据库或文件系统),以实现数据同步和数据备份等功能，并将其转换为DataStream流。然后，Sink将这个DataStream流写入到外部系统中，以便举行后续的数据处理和分析。

1. import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
2. public class CustomSink extends RichSinkFunction<String> {
4.     @Override
5.     public void invoke(String value, Context context) throws Exception {
6.         System.out.println("json->: "+value);
7.     }
8. }

复制代码

这段代码界说了一个名为CustomSink的类，它继承自RichSinkFunction类。RichSinkFunction是Flink CDC中用于将数据写入外部系统(如数据库或文件系统)的函数接口。CustomSink的作用是将CDC吸收到的数据写入外部系统中。详细实现方式由子类CustomSink来界说。由于这个类继承了RichSinkFunction,因此可以利用Flink中的其他RichSink函数特性，比方设置日志级别、设置连接等,invoke则是处理函数。
main设置运行

如下面的代码，构建Flink CDC连接

1.   public static void main(String[] args) throws Exception {
2.         MySqlSourceBuilder<String> builder = MySqlSource.builder();
3.         MySqlSource<String> source = builder.hostname("192.168.2.6")
4.                 .port(3306)
5.                 .databaseList("mydb")
6.                 .tableList("mydb.user")
7.                 .username("root")
8.                 .password("root")
9.                 .deserializer(new JsonDebeziumDeserializationSchema())
10.                 .includeSchemaChanges(true)
11.                 .build();
12. //        启动webui，绑定本地web-ui端口号
13.         Configuration configuration=new Configuration();
14.         configuration.setInteger(RestOptions.PORT,8081);
15.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
17.         env.enableCheckpointing(5000);
18.         env.fromSource(source, WatermarkStrategy.noWatermarks(),"MYSQL Source")
19.                 .addSink(new CustomSink());
20.         env.execute();
21.     }

复制代码

这段代码是利用Flink构建一个数据流处理使命，从MySQL数据库中读取数据并举行处理。
首先，利用MySqlSourceBuilder创建一个MySqlSource对象，并设置连接参数(hostname、port、databaseList、tableList、username和password)以及反序列化器(JsonDebeziumDeserializationSchema)。然后，创建Configuration对象并设置WebUI端口号(RestOptions.PORT),接着利用StreamExecutionEnvironment创建一个执行环境，启用检查点(checkpointing)并将MySqlSource和自界说的Sink添加到执行环境中。末了，执行整个使命。
操作数据库查看结果

如图所示：

1. {
2.   "before": null,
3.   "after": {
4.     "id": "1661935564737286146",
5.     "qu_type": 4,
6.     "level": 1,
7.     "image": "",
8.     "content": "dos查看日期、时间",
9.     "create_time": 1685100091000,
10.     "update_time": 1685100091000,
11.     "remark": "",
12.     "analysis": ""
13.   },
14.   "source": {
15.     "version": "1.6.4.Final",
16.     "connector": "mysql",
17.     "name": "mysql_binlog_source",
18.     "ts_ms": 0,
19.     "snapshot": "false",
20.     "db": "mydb",
21.     "sequence": null,
22.     "table": "t_user",
23.     "server_id": 0,
24.     "gtid": null,
25.     "file": "",
26.     "pos": 0,
27.     "row": 0,
28.     "thread": null,
29.     "query": null
30.   },
31.   "op": "r",
32.   "ts_ms": 1685521801276,
33.   "transaction": null
34. }

复制代码

这段代码是Flink CDC监听到的MySQL数据库binlog中的一条变动数据记录，表示在MySQL的t_user表中发生了一次读取操作，读取的数据记录的内容为：

• “id”: “1661935564737286146”
  
• “qu_type”: 4
  
• “level”: 1
  
• “image”: “”
  
• “content”: “dos查看日期、时间”
  
• “create_time”: 1685100091000
  
• “update_time”: 1685100091000
  
• “remark”: “”
  
• “analysis”: “” 此中，before字段为null，表示这是一条insert操作；after字段为变动后的数据内容；source字段表示数据的来源信息，包括MySQL的版本、连接器范例、数据库名称、表名称等；op字段表示操作范例，"r"表示读取操作；ts_ms字段表示变动发生的时间戳；transaction字段表示事件信息，这里为null，表示这是一条非事件性的操作记录。
  

操作数据JSON讲解

在Flink CDC中，op字段表示MySQL数据库binlog中的操作范例，通常环境下分为以下几种范例：

• “c”：表示create，表示对数据库举行了创建操作。
  
• “u”：表示update，表示对数据库举行了更新操作。
  
• “d”：表示delete，表示对数据库举行了删除操作。
  
• “r”：表示read，表示对数据库举行了读取操作。 此中，前三种操作范例都是数据的变动操作，read操作则是指对数据库举行的查询操作。在Flink CDC中，一样平常只会监听到前三种操作范例，因为只有这三种操作范例才会导致数据库中的数据发生变革，而read操作则只是查询数据，并不会导致数据的变革。
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。