Flink CDC介绍

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

<properties>
	<flink.version>1.13.0</flink.version>
	<maven.compiler.source>1.8</maven.compiler.source>
	<maven.compiler.target>1.8</maven.compiler.target>
</properties>
<dependency>
	<groupId>com.ververica</groupId>
	<artifactId>flink-connector-mysql-cdc</artifactId>
	<version>${flinkcdc.version}</version>
</dependency>
<!-- flink核心API -->
<dependency>
	<groupId>org.apache.flink</groupId>
	<artifactId>flink-clients_2.12</artifactId>
	<version>${flink.version}</version>
</dependency>
<dependency>
	<groupId>org.apache.flink</groupId>
	<artifactId>flink-java</artifactId>
	<version>${flink.version}</version>
</dependency>
<dependency>
	<groupId>org.apache.flink</groupId>
	<artifactId>flink-scala_2.12</artifactId>
	<version>${flink.version}</version>
</dependency>
<dependency>
	<groupId>org.apache.flink</groupId>
	<artifactId>flink-streaming-java_2.12</artifactId>
	<version>${flink.version}</version>
</dependency>
<dependency>
	<groupId>org.apache.flink</groupId>
	<artifactId>flink-streaming-scala_2.12</artifactId>
	<version>${flink.version}</version>
</dependency>
<dependency>
	<groupId>org.apache.flink</groupId>
	<artifactId>flink-table-common</artifactId>
	<version>${flink.version}</version>
</dependency>
<dependency>
	<groupId>org.apache.flink</groupId>
	<artifactId>flink-table-planner-blink_2.12</artifactId>
	<version>${flink.version}</version>
</dependency>
<dependency>
	<groupId>org.apache.flink</groupId>
	<artifactId>flink-table-api-java-bridge_2.12</artifactId>
	<version>${flink.version}</version>
</dependency>

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

my.cnf

[client]
default_character_set=utf8
[mysqld]
server-id=1
collation_server=utf8_general_ci
character_set_server=utf8
log-bin=mysql-bin
binlog_format=row
expire_logs_days=30

ddl&dml.sql

create table test_cdc
(
    id   int          not null
        primary key,
    name varchar(100) null,
    age  int          null
);

INSERT INTO flink.test_cdc (id, name, age) VALUES (1, 'Daniel', 25);
INSERT INTO flink.test_cdc (id, name, age) VALUES (2, 'David', 38);
INSERT INTO flink.test_cdc (id, name, age) VALUES (3, 'James', 16);
INSERT INTO flink.test_cdc (id, name, age) VALUES (4, 'Robert', 27);

FlinkDSCDC.java

package com.daniel.util;

import com.ververica.cdc.connectors.mysql.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.DebeziumSourceFunction;
import com.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author Daniel
 * @Date: 2023/7/25 10:03
 * @Description DataStream API CDC
 **/
public class FlinkDSCDC {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DebeziumSourceFunction<String> sourceFunction = MySqlSource.<String>builder()
                .hostname("localhost")
                .port(3306)
                .username("root")
                .password("123456")
                .databaseList("flink")
                // 这里一定要是db.table的形式
                .tableList("flink.test_cdc")
                .deserializer(new StringDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial())
                .build();

        DataStreamSource<String> dataStreamSource = env.addSource(sourceFunction);
        dataStreamSource.print();
        env.execute("FlinkDSCDC");
    }
}


UPDATE flink.test_cdc t SET t.age = 24 WHERE t.id = 1;
UPDATE flink.test_cdc t SET t.name = 'Andy' WHERE t.id = 3;

打印出的日志

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=)}
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

package com.daniel.util;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @Author Daniel
 * @Date: 2023/7/25 15:25
 * @Description
 **/
public class FlinkSQLCDC {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.executeSql("CREATE TABLE test_cdc (" +
                " id int primary key," +
                " name STRING," +
                " age int" +
                ") WITH (" +
                " 'connector' = 'mysql-cdc'," +
                " 'scan.startup.mode' = 'latest-offset'," +
                " 'hostname' = 'localhost'," +
                " 'port' = '3306'," +
                " 'username' = 'root'," +
                " 'password' = '123456'," +
                " 'database-name' = 'flink'," +
                " 'table-name' = 'test_cdc'" +
                ")");

        Table table = tableEnv.sqlQuery("select * from test_cdc");
        DataStream<Tuple2<Boolean, Row>> dataStreamSource = tableEnv.toRetractStream(table, Row.class);
        dataStreamSource.print();
        env.execute("FlinkSQLCDC");
    }
}


UPDATE flink.test_cdc t SET t.age = 55 WHERE t.id = 2;
UPDATE flink.test_cdc t SET t.age = 22 WHERE t.id = 3;
UPDATE flink.test_cdc t SET t.name = 'Alice' WHERE t.id = 4;
UPDATE flink.test_cdc t SET t.age = 18 WHERE t.id = 1;
INSERT INTO flink.test_cdc (id, name, age) VALUES (5, 'David', 29);

打印出的日志

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

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mfbz.cn/a/95643.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

PCL common模块应用实例

PCL common模块应用实例

目录 一、common模块中的头文件二、基本函数1、angles.h2、centriod.h3、common.h4、distance.h5、copy_point.h6、geometry.h参考链接本文由CSDN点云侠原创，
阅读更多...
Excel：如何实现分组内的升序和降序？

Excel：如何实现分组内的升序和降序？

一、POWER 1、构建辅助列D列&#xff0c;在D2单元格输入公式&#xff1a; -POWER(10,COUNTA($A$2:A2)3)C2 2、选中B1:D10&#xff0c;注意不能宣导A列的合并单元格&#xff0c;进行以下操作&#xff1a; 3、删除辅助列即可 二、COUNTA 第一步&#xff0c;D2建立辅助列&#xf…
阅读更多...
day 6 c++

day 6 c++

#include <iostream>using namespace std; class Animal { public:Animal(){cout << "训练员的无参构造" << endl;}virtual void perform(){} }; class Tiger:public Animal {string tezheng;string biaoyan; public:Tiger(){cout << "…
阅读更多...
财务部发布《企业数据资源相关会计处理暂行规定》

财务部发布《企业数据资源相关会计处理暂行规定》

导读 财务部为规范企业数据资源相关会计处理&#xff0c;强化相关会计信息披露&#xff0c;根据《中华人民共和国会计法》和相关企业会计准则&#xff0c;制定了《企业数据资源相关会计处理暂行规定》。 加gzh“大数据食铁兽”&#xff0c;回复“20230828”获取材料完整版 来…
阅读更多...
对class文件进行base64编码

对class文件进行base64编码

使用以下代码 package org.springframework.cloud.gateway.sample;import org.springframework.util.Base64Utils;import java.io.*; import java.nio.charset.StandardCharsets;public class EncodeShell {public static void main(String[] args){byte[] data null;try {In…
阅读更多...
2024王道408数据结构P144 T18

2024王道408数据结构P144 T18

2024王道408数据结构P144 T18 思考过程 首先还是先看题目的意思&#xff0c;让我们在中序线索二叉树里查找指定结点在后序的前驱结点&#xff0c;这题有一点难至少对我来说…我讲的不清楚理解一下我做的也有点糊涂。在创建结构体时多两个变量ltag和rtag&#xff0c;当ltag0时…
阅读更多...
基于stm32的ADS1292R 心电波形采集

基于stm32的ADS1292R 心电波形采集

一、前言 ADS1292R是TI公司早在几年前出产的一款医用级ADC芯片&#xff0c;它主要应用在医疗仪器(心电图ECG),可以监护患者以及病人护理和健身监视器。ADS1292R集成了心电采集所需要的部件&#xff0c;方便设备小型化。它的功耗极低&#xff0c;使得可以作为长时间监控成为可能…
阅读更多...
在vue3项目中编辑的时候，解决对话框里边的数据和列表中的数据联动了。深复制

在vue3项目中编辑的时候，解决对话框里边的数据和列表中的数据联动了。深复制

//分析原因是从列表中拿到的数据直接复制去修改就涉及到堆里变的内容是一样的&#xff0c;直接复制其实只是把引用地址赋值给变量了&#xff0c;解决方法是 浅复制和深复制。<!-- 审批流程管理 --> <template><div style"float: left; width: 250px;backgr…
阅读更多...
yolov8使用C++推理的流程及注意事项

yolov8使用C++推理的流程及注意事项

1.下载yolov8项目源码GitHub - ultralytics/ultralytics: NEW - YOLOv8 &#x1f680; in PyTorch > ONNX > OpenVINO > CoreML > TFLite 2.下载opencvReleases - OpenCV,建议版本>4.7.0,选择下载源码&#xff0c; windows版本由于使用的编译器与我们所使用的m…
阅读更多...
DP读书：鲲鹏处理器 架构与编程（十三）操作系统内核与云基础软件

DP读书：鲲鹏处理器 架构与编程（十三）操作系统内核与云基础软件

操作系统内核与云基础软件 鲲鹏软件构成硬件特定软件 鲲鹏软件构成硬件特定软件1. Boot Loader2. SBSA 与 SBBR3. UEFI4. ACPI 操作系统内核Linux系统调用Linux进程调度Linux内存管理Linux虚拟文件系统Linux网络子系统Linux进程间通信Linux可加载内核模块Linux设备驱动程序Linu…
阅读更多...
open cv快速入门系列---数字图像基础

open cv快速入门系列---数字图像基础

目录 一、数字图像基础 1.1 数字图像和图像单位 1.2 区分图片分辨率与屏幕分辨率 1.3 图像的灰度与灰度级 1.4 图像的深度 1.5 二值图像、灰度图像与彩色图像 1.6 通道数 二、数字图像处理 2.1 图像噪声及其消除 2.2 数字图像处理技术 2.2.1 图像变换 2.2.2 图像增强…
阅读更多...
同创永益入选首批“金融数字韧性与混沌工程实践试点机构”

同创永益入选首批“金融数字韧性与混沌工程实践试点机构”

8月16日下午&#xff0c;由北京国家金融科技认证中心、北京国家金融标准化研究院联合主办的“传递信任 服务发展”金融科技标准认证生态大会在太原成功举办。中国金融电子化集团有限公司党委书记、董事长周逢民&#xff0c;中国科学院院士冯登国&#xff0c;中国工商银行首席技…
阅读更多...
哲讯科技携手无锡华启动SCM定制化项目，共谋数字化转型之路

哲讯科技携手无锡华启动SCM定制化项目，共谋数字化转型之路

无锡华光座椅弹簧有限公司启动SCM定制化项目 近日&#xff0c;无锡华光座椅弹簧有限公司顺利举行了SCM定制化项目的启动会。本次启动会作为该项目实施的重要里程碑&#xff0c;吸引了双方项目组核心成员的共同参与&#xff0c;并见证了项目的正式启动。 无锡华光座椅弹簧有限公…
阅读更多...
uni-app开发小程序，radio单选按钮，点击可以选中，再次点击可以取消

uni-app开发小程序，radio单选按钮，点击可以选中，再次点击可以取消

一、实现效果&#xff1a; 二、代码实现&#xff1a; 不适用官方的change方法&#xff0c;自己定义点击方法。 动态判断定义的值是否等于遍历的值进行回显&#xff0c;如果和上一次点击的值一样&#xff0c;就把定义的值改为null <template><view><radio-group&…
阅读更多...
kafka调优配置

kafka调优配置

Kafka生产者核心参数配置 来源于尚硅谷 参数名称描述bootstrap.servers生产者连接集群所需的broker地址清单。例如hadoop102:9092,hadoop103:9092,hadoop104:9092&#xff0c;可以设置1个或者多个&#xff0c;中间用逗号隔开。注意这里并非需要所有的broker地址&#xff0c;因…
阅读更多...
Linux 常用命令 2

Linux 常用命令 2

Linux 常用命令 2 1、组和权限管理1.1、ls 指令1.2、chown 指令1.3、chgrp 指令1.4、chmod 指令1.5、chown 指令1.6、chgrp 指令 2、crond 任务调度2.1、crontab2.2、时间格式2.3、脚本无法执行问题2.4、案例 3、进程管理3.1、ps 指令3.2、kill 和 killall 指令3.3、pstree 指令…
阅读更多...
QSqlDatabase（2）实例，QTableView显示数据库表数据

QSqlDatabase（2）实例，QTableView显示数据库表数据

目录 前言 1、实现的功能 2、具体的代码实现 前言 想了解QSqlDatabase基本知识的&#xff0c;以及增删改查的用法&#xff0c;可以浏览上一篇文章&#xff1a; QSqlDatabase&#xff08;1&#xff09;基本接口&#xff0c;以及(增删改除)的简单实例_Ivy_belief的博客-CSDN…
阅读更多...
电脑识别不了固态硬盘怎么办？

电脑识别不了固态硬盘怎么办？

在使用固态硬盘时&#xff0c;可能会出现电脑无法识别的情况&#xff0c;这时我们就无法使用固态硬盘中的数据。那么&#xff0c;电脑识别不了固态硬盘怎么办&#xff1f; 为什么电脑识别不了固态硬盘&#xff1f; 一般来说&#xff0c;电脑识别不了固态硬盘是因为以下3个原因…
阅读更多...
HTML及CSS入门及精通

HTML及CSS入门及精通

前言 HTML&#xff08;超文本标记语言&#xff09;和CSS&#xff08;层叠样式表&#xff09;是构建网页的两个基本技术。HTML用于定义网页的结构和内容&#xff0c;而CSS用于控制网页的样式和布局。本教程将介绍HTML和CSS的入门知识&#xff0c;并逐步引导您掌握更高级的技巧和…
阅读更多...
css元素定位：通过元素的标签或者元素的id、class属性定位

css元素定位：通过元素的标签或者元素的id、class属性定位

前言 大部分人在使用selenium定位元素时&#xff0c;用的是xpath元素定位方式&#xff0c;因为xpath元素定位方式基本能解决定位的需求。xpath元素定位方式更直观&#xff0c;更好理解一些。 css元素定位方式往往被忽略掉了&#xff0c;其实css元素定位方式也有它的价值&…
阅读更多...
最新文章

Copyright @ 2022~2023

友情链接： 我的编程经验分享 一条长河网 尧图网站开发 尧图建网站