当前位置：首页 > news >正文

一级a视网站做爰片公司网站建设阜阳

news 2025/9/27 15:14:56

一级a视网站做爰片,公司网站建设阜阳,做影视网站什么cms好用吗,阿里云备案网站建设方案书一、什么是 CDCCDC 是 Change Data Capture(变更数据获取)的简称。核心思想是，监测数据库的bianeng并捕获数据库的变动（包括数据或数据表的插入、更新以及删除等），将这些变更按发生的顺序完整记录下来，写入到消息中间件…

一、什么是 CDC

CDC 是 Change Data Capture(变更数据获取)的简称。核心思想是，监测数据库的bia'neng并捕获数据库的变动（包括数据或数据表的插入、更新以及删除等），将这些变更按发生的顺序完整记录下来，写入到消息中间件中以供其他服务进行订阅及消费。

二、Flink CDC

Flink CDC 通过捕获数据库的变更日志（如 MySQL binlog、Postgres WAL），实现高效、低延迟的数据同步。其核心工作原理可分为变更数据捕获、转换处理和输出同步三个阶段：

2.1 变更数据捕获

日志抓取：

使用 Debezium (Flink CDC 底层引擎) 连接数据库，通过各自协议读取变更日志

MySQL ——> binlog
PostgreSQL ——> WAL
Oracle ——> Redo Log
初始快照：

首次启动时执行全量快照（snapshot），将表中已有数据转为 INSERT 事件流。

采用分块并行读取（如按主键分片）加速同步。

2.2 转换处理

日志解析与转换

格式转换：

Debezium 将原始二进制日志解析为 JSON / Avro 结构，Flink CDC将其转换为 RowData结构（Flink 内部数据结构）。

Schema 映射：

自动推断表结构（字段名、类型、主键），动态处理 DDL 变更（如新增列）。
Flink 实时处理

流式处理：

变更事件（INSERT / UPDATE / DELETE）作为无界数据流进入 Flink。

Exactly-Once 语义：

通过 Checkpoint 机制 + binlog 位点持久化（如 Kafka 或 Flink State）保证数据一致性。

转换能力：

支持使用 Flink SQL 或 DataStream API 进行过滤、聚合、关联维表等操作。

2.3 输出同步

将处理后的数据写入下游系统：

数据胡：Hudi / IceBerg
OLAP 引擎：Clickhouse / Doris
消息队列：Kafka / Pulsar
数据库：Mysql / PostgreSQL（需要支持 Upsert）

三、关键技术

3.1 全量+增量无缝切换

无锁读取：

快照阶段使用 SELECT ... FROM 而非锁表（Mysql 使用 mysqldump 的轻量模式）。
断点续传：

Checkpoint 存储 binlog 位点，故障恢复时从断点继续同步。

3.2 动态表结构处理

自动 Schema 同步：

源表新增列时，Flink CDC 自动更新 Schema，下游系统需支持 DDL 传播。
兼容性处理：

旧数缺失实新增列时填充 NULL。

3.3 并行读取优化

分片策略：

按主键范围或按时间分区并行快照。
增量阶段并行度：

单任务串行读取 binlog（避免乱序），但可并行处理后续计算。

四、典型应用场景

4.1 实时数仓同步

-- Flink SQL 实现 MySQL → Hudi
CREATE TABLE orders_cdc (id BIGINT,amount DECIMAL(10,2),PRIMARY KEY (id) NOT ENFORCED
) WITH ('connector' = 'mysql-cdc','hostname' = 'mysql-host','database-name' = 'test','table-name' = 'orders'
);CREATE TABLE hudi_orders (...) WITH ('connector'='hudi');INSERT INTO hudi_orders SELECT * FROM orders_cdc;

4.2 多源数据聚合

// DataStream 实现订单+用户表关联
DataSource<Order> orders = env.fromSource(MySqlSource.<Order>builder().build(), ...);DataSource<User> users = env.fromSource(PostgresSource.<User>builder().build(), ...);orders.connect(users).keyBy(o -> o.userId, u -> u.id).process(new EnrichOrderFunction()); // 关联用户信息