关于flink:基于-Flink-SQL-CDC-的实时数据同步方案

作者：伍翀 (云邪)

整顿：陈政羽（Flink 社区志愿者）

Flink 1.11 引入了 Flink SQL CDC，CDC 能给咱们数据和业务间能带来什么变动？本文由 Apache Flink PMC，阿里巴巴技术专家伍翀 (云邪）分享，内容将从传统的数据同步计划，基于 Flink CDC 同步的解决方案以及更多的利用场景和 CDC 将来开发布局等方面进行介绍和演示。

1、传统数据同步计划

2、基于 Flink SQL CDC 的数据同步计划（Demo）

3、Flink SQL CDC 的更多利用场景

4、Flink SQL CDC 的将来布局

直播回顾：
https://www.bilibili.com/video/BV1zt4y1D7kt/

业务零碎常常会遇到须要更新数据到多个存储的需要。例如：一个订单零碎刚刚开始只须要写入数据库即可实现业务应用。某天 BI 团队冀望对数据库做全文索引，于是咱们同时要写多一份数据到 ES 中，革新后一段时间，又有需要须要写入到 Redis 缓存中。

很显著这种模式是不可继续倒退的，这种双写到各个数据存储系统中可能导致不可保护和扩大，数据一致性问题等，须要引入分布式事务，老本和复杂度也随之减少。咱们能够通过 CDC（Change Data Capture）工具进行解除耦合，同步到上游须要同步的存储系统。通过这种形式进步零碎的稳健性，也不便后续的保护。

CDC 全称是 Change Data Capture，它是一个比拟狭义的概念，只有能捕捉变更的数据，咱们都能够称为 CDC。业界次要有基于查问的 CDC 和基于日志的 CDC，能够从上面表格比照他们性能和差别点。

通过以上比照，咱们能够发现基于日志 CDC 有以下这几种劣势：

· 可能捕捉所有数据的变动，捕捉残缺的变更记录。在异地容灾，数据备份等场景中失去广泛应用，如果是基于查问的 CDC 有可能导致两次查问的两头一部分数据失落

· 每次 DML 操作均有记录无需像查问 CDC 这样发动全表扫描进行过滤，领有更高的效率和性能，具备低提早，不减少数据库负载的劣势

· 无需入侵业务，业务解耦，无需更改业务模型

· 捕捉删除事件和捕捉旧记录的状态，在查问 CDC 中，周期的查问无奈感知两头数据是否删除

从 ETL 的角度进行剖析，个别采集的都是业务库数据，这里应用 MySQL 作为须要采集的数据库，通过 Debezium 把 MySQL Binlog 进行采集后发送至 Kafka 音讯队列，而后对接一些实时计算引擎或者 APP 进行生产后把数据传输入 OLAP 零碎或者其余存储介质。

Flink 心愿买通更多数据源，施展残缺的计算能力。咱们生产中次要来源于业务日志和数据库日志，Flink 在业务日志的反对上曾经十分欠缺，然而在数据库日志反对方面在 Flink 1.11 前还属于一片空白，这就是为什么要集成 CDC 的起因之一。

Flink SQL 外部反对了残缺的 changelog 机制，所以 Flink 对接 CDC 数据只须要把 CDC 数据转换成 Flink 意识的数据，所以在 Flink 1.11 外面重构了 TableSource 接口，以便更好反对和集成 CDC。

重构后的 TableSource 输入的都是 RowData 数据结构，代表了一行的数据。在 RowData 下面会有一个元数据的信息，咱们称为 RowKind。RowKind 外面包含了插入、更新前、更新后、删除，这样和数据库外面的 binlog 概念非常相似。通过 Debezium 采集的 JSON 格局，蕴含了旧数据和新数据行以及原数据信息，op 的 u 示意是 update 更新操作标识符，ts_ms 示意同步的工夫戳。因而，对接 Debezium JSON 的数据，其实就是将这种原始的 JSON 数据转换成 Flink 意识的 RowData。

当抉择 Flink 作为 ETL 工具时，在数据同步场景，如下图同步构造：

通过 Debezium 订阅业务库 MySQL 的 Binlog 传输至 Kafka，Flink 通过创立 Kafka 表指定 format 格局为 debezium-json，而后通过 Flink 进行计算后或者直接插入到其余内部数据存储系统，例如图中的 Elasticsearch 和 PostgreSQL。

然而这个架构有个毛病，咱们能够看到采集端组件过多导致保护繁冗，这时候就会想是否能够用 Flink SQL 间接对接 MySQL 的 binlog 数据呢，有没能够代替的计划呢？

答案是有的！通过改良后构造如下图：

社区开发了 flink-cdc-connectors 组件，这是一个能够间接从 MySQL、PostgreSQL 等数据库间接读取全量数据和增量变更数据的 source 组件。目前也已开源，开源地址：

https://github.com/ververica/flink-cdc-connectors

flink-cdc-connectors 能够用来替换 Debezium+Kafka 的数据采集模块，从而实现 Flink SQL 采集 + 计算 + 传输（ETL）一体化，这样做的长处有以下：

· 开箱即用，简略易上手

· 缩小保护的组件，简化实时链路，加重部署老本

· 减小端到端提早

· Flink 本身反对 Exactly Once 的读取和计算

· 数据不落地，缩小存储老本

· 反对全量和增量流式读取

· binlog 采集位点可回溯 *

上面给大家带来 3 个对于 Flink SQL + CDC 在理论场景中应用较多的案例。在实现试验时候，你须要 Docker、MySQL、Elasticsearch 等组件，具体请参考每个案例参考文档。

这个案例通过订阅咱们订单表（事实表）数据，通过 Debezium 将 MySQL Binlog 发送至 Kafka，通过维表 Join 和 ETL 操作把后果输入至上游的 PG 数据库。具体能够参考 Flink 公众号文章：《Flink JDBC Connector：Flink 与数据库集成最佳实际》案例进行实际操作。

https://www.bilibili.com/video/BV1bp4y1q78d

模仿电商公司的订单表和物流表，须要对订单数据进行统计分析，对于不同的信息须要进行关联后续造成订单的大宽表后，交给上游的业务方应用 ES 做数据分析，这个案例演示了如何只依赖 Flink 不依赖其余组件，借助 Flink 弱小的计算能力实时把 Binlog 的数据流关联一次并同步至 ES。

例如如下的这段 Flink SQL 代码就能实现实时同步 MySQL 中 orders 表的全量 + 增量数据的目标。

CREATE TABLE orders (
  order_id INT,
  order_date TIMESTAMP(0),
  customer_name STRING,
  price DECIMAL(10, 5),
  product_id INT,
  order_status BOOLEAN
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = 'localhost',
  'port' = '3306',
  'username' = 'root',
  'password' = '123456',
  'database-name' = 'mydb',
  'table-name' = 'orders'
);

SELECT * FROM orders

为了让读者更好地上手和了解，咱们还提供了 docker-compose 的测试环境，更具体的案例教程请参考下文的视频链接和文档链接。

视频链接：
https://www.bilibili.com/video/BV1zt4y1D7kt
文档教程：
https://github.com/ververica/flink-cdc-connectors/wiki/ 中文教程

上面案例就是对 GMV 进行天级别的全站统计。蕴含插入 / 更新 / 删除，只有付款的订单能力计算进入 GMV，察看 GMV 值的变动。

视频链接：
https://www.bilibili.com/video/BV1zt4y1D7kt
文档教程：
https://github.com/ververica/flink-cdc-connectors/wiki/ 中文教程

Flink SQL CDC 不仅能够灵便地利用于实时数据同步场景中，还能够买通更多的场景提供给用户抉择。

· 如果你曾经有 Debezium/Canal + Kafka 的采集层 (E)，能够应用 Flink 作为计算层 (T) 和传输层 (L)

· 也能够用 Flink 代替 Debezium/Canal，由 Flink 间接同步变更数据到 Kafka，Flink 对立 ETL 流程

· 如果不须要 Kafka 数据缓存，能够由 Flink 间接同步变更数据到目的地，Flink 对立 ETL 流程

· 实时数据同步，数据备份，数据迁徙，数仓构建

劣势：丰盛的上下游（E & L），弱小的计算（T），易用的 API（SQL），流式计算低提早

· 数据库之上的实时物化视图、流式数据分析

· 索引构建和实时保护

· 业务 cache 刷新

· 审计跟踪

· 微服务的解耦，读写分

· 基于 CDC 的维表关联

上面介绍一下为何用 CDC 的维表关联会比基于查问的维表查问快。

■ 基于查问的维表关联

目前维表查问的形式次要是通过 Join 的形式，数据从音讯队列进来后通过向数据库发动 IO 的申请，由数据库把后果返回后合并再输入到上游，然而这个过程无可避免的产生了 IO 和网络通信的耗费，导致吞吐量无奈进一步晋升，就算应用一些缓存机制，然而因为缓存更新不及时可能会导致精确性也没那么高。

■ 基于 CDC 的维表关联

咱们能够通过 CDC 把维表的数据导入到维表 Join 的状态外面，在这个 State 外面因为它是一个分布式的 State，外面保留了 Database 外面实时的数据库维表镜像，当音讯队列数据过去时候无需再次查问近程的数据库了，间接查问本地磁盘的 State，防止了 IO 操作，实现了低提早、高吞吐，更精准。

Tips：目前此性能在 1.12 版本的布局中，具体进度请关注 FLIP-132。

· FLIP-132：Temporal Table DDL（基于 CDC 的维表关联）

· Upsert 数据输入到 Kafka

· 更多的 CDC formats 反对（debezium-avro, OGG, Maxwell）

· 批模式反对解决 CDC 数据

· flink-cdc-connectors 反对更多数据库

本文通过比照传统的数据同步计划与 Flink SQL CDC 计划分享了 Flink CDC 的劣势，与此同时介绍了 CDC 分为日志型和查问型各自的实现原理。后续案例也演示了对于 Debezium 订阅 MySQL Binlog 的场景介绍，以及如何通过 flink-cdc-connectors 实现技术整合代替订阅组件。除此之外，还具体解说了 Flink CDC 在数据同步、物化视图、多机房备份等的场景，并重点解说了社区将来布局的基于 CDC 维表关联比照传统维表关联的劣势以及 CDC 组件工作。

心愿通过这次分享，大家对 Flink SQL CDC 能有全新的意识和理解，在将来理论生产开发中，冀望 Flink CDC 能带来更多开发的便捷和更丰盛的应用场景。

1、GROUP BY 后果如何写到 Kafka？

因为 group by 的后果是一个更新的后果，目前无奈写入 append only 的音讯队列中外面去。更新的后果写入 Kafka 中将在 1.12 版本中原生地反对。在 1.11 版本中，能够通过 flink-cdc-connectors 我的项目提供的 changelog-json format 来实现该性能，具体见文档。

文档链接：
https://github.com/ververica/flink-cdc-connectors/wiki/Changelog-JSON-Format

2、CDC 是否须要保障程序化生产？

是的，数据同步到 kafka，首先须要 kafka 在分区中保障有序，同一个 key 的变更数据须要打入到同一个 kafka 的分区外面。这样 flink 读取的时候能力保障程序。

关于flink:基于-Flink-SQL-CDC-的实时数据同步方案

传统的数据同步计划与 Flink SQL CDC 解决方案

Flink SQL CDC 数据同步与原理解析

基于日志的 CDC 计划介绍

抉择 Flink 作为 ETL 工具

基于 Flink SQL CDC 的数据同步计划实际

案例 1 : Flink SQL CDC + JDBC Connector

案例 2 : CDC Streaming ETL

案例 3 : Streaming Changes to Kafka

Flink SQL CDC 的更多利用场景

Flink 在数据同步场景中的灵便定位

Flink SQL CDC : 买通更多场景

将来布局

总结

Q & A