Amoro + Flink CDC 数据融合入湖新体验

摘要：本文整理自货拉拉高级大数据开发工程师，Apache Amoro PMC 陈政羽老师，在 Flink Forward Asia 2024 数据集成（一）专场的分享。内容分为以下四个部分：

1、CDC 在货拉拉应用

2、数据入湖新体验

3、入湖优化

4、未来规划

01.Flink CDC 在货拉拉应用

首先讲解 Flink CDC 目前在货拉拉上的应用以及场景。

1.1 Flink CDC 使用量

CDC 是上半年开始接入的数据集成方案，现在已经有 50 多个任务跑在正式生产环境上。我们希望后续建设一个标准化的数据采集平台和数据同步的平台，将后续比较老旧的任务 canal 取消。目前数据量每天都在 TB 级以上，包括一些订单和司机的数据。我们还进行了一些分库分表的采集，基本每一个采集都包含几千张表进行迁表迁库的同步。

1.2 落地场景

目前落地的一些场景如下。我们公司目前有几个大的业务例如货拉拉、小拉出行和海外的业务 LaLaMove 和跑腿等业务线。公司的业务不仅仅在国内还在海外，所以还会有多 DC 的环境，整体数据业务采集量达到 TB-PB 级别。目前接入的 CDC 核心业务例如云台、实时看板、kepler、实时报表以及交易等业务。我们希望基于 Flink CDC 3.0 以上版本实现整个数据链路的以旧换新，进行数据链路的替换工作，同时对整个数据采集进行平台化工作。

1.3 稳定性建设

在落地时遇到了一些稳定性的挑战。从稳定性上进行多维度方面建设。首先从应用上层上有一个实时计算平台飞流，通过封装飞流的一些任务，将这部分订阅 SQL 化，提供给业务方使用。后续希望基于 3.0 yaml 的一些特性封装给业务方进行使用，业务方不需要对 Flink SQL 有太多了解，通过在网页上配置和审批的方式就可以发布实时数据订阅的动作。在平台上适配也做了一些工作，例如感知能力的对齐、做一些 SDK，因为 CDC 不仅是大数据库在使用，还有一些业务方也在使用 CDC 的数据订阅，所以需要封装一套通用的 SDK 给业务方做下游的数据消费。同时在稳定性上也做了一些工作，例如 HA、限流、性能验收等测试工作。

1.4 数据采集入湖场景

下面讲解基于以上工作后下一步希望推进的工作。CDC 是一个 Changelog 订阅数据库的 Binlog 采集类的一些事件。在一些数据采集的入湖场景中，以下列举三个入湖场景，不仅限于 CDC。首先在公司内部场景有一些埋点上报，这部分数据对于时效性要求较低，同时是非结构化数据。CDC 是一种数据入湖的分析，基于 Binlog，有 Upsert、Insert、Delesert 操作，业务方对于此处的数据时效性要求较高。一般从关系型数据库采集，所以一般都是一些结构化的数据。像一些日志数据的入湖分析，需要间接进行一些统计。所以主要针对以上三种场景，列出三种计算。第一种就是离线计算，吞吐量较高，但是响应性低，一般都是经过一些行列的读取，存储周期较长。另外增量计算可以在延迟和吞吐上获得一定平衡，但是存储周期也比较短或者可能比较中等去做一些间接性的中等计算。Flink 计算进行一些低延迟、高响应的动作 。

以上讲解了一些 CDC 入湖的场景，在 CDC 上入湖会遇到一些什么挑战呢？下面介绍当前 CDC 3.0 版本或 paimon 等一些新特性可能带来的一些挑战。

02.CDC 入湖新体验

2.1 CDC 3.0 YAML With Paimon

首先在 3.0 上已经提到，之前是基于 Cdc Connector 开发，没有基于 Pipeline。3.0 版本已经提供基于 Pipeline 的形式，还有 Route、Source、Sink、Transform、Schema Evolution。这些在 Paimon 上都具有，但是我们希望在一些离线链路场景上还能够支持 Iceberg 能力。我们会基于 Iceberg Flink Connector v2 去开发 Iceberg Flink Connector v2 的 Pipeline。

2.2 数据入湖困痛点-文件碎片化严重

下面讲解目前数据入湖比较大的痛点——文件碎片化严重。下图在 Paimon 网站上摘录。CDC 数据在进行一些入湖时可以看到，数据有原数据、真实数据包括 Manifest、datafile，这样就会导致原数据以及 datafile 有表多的问题。特别是在一些场景中例如数据的上报延迟，就会导致很多小文件写入到不同的分区上，给 Hdfs 的压力增大。同时在对象存储上访问频率增加，会导致一些作业存在性能瓶颈。Paimon 会通过自己的自动优化机制定期进行一些小文件合并，但是也存在一些问题，例如在进行小文件合并时对用户而言是黑盒的操作，没有办法控制合并频率、生命周期或者合并的时间。

2.3 数据入湖痛点-Schema 演进

第二点在 CDC 入湖时会有一些 Schema 演进，这种 Schema 的演进会产生一些问题，例如 Update_table 的一些操作。在下游同步这些 Schema 时是否也要将这部分 Schema 同步呢？例如 Update_table 某一个字段，修改完某一个字段后业务也要对应的感知，下流消费的 Flink SQL 也需要对应进行改变。可以通过一些手段减缓这种影响，避免作业 fell over。这里建议大家对 Schema Evolution 做一些合理配置过滤掉实时写入到湖仓中不需要的 Schema Evolution。例如高危操作 drop table 或者 Update_table 发生一些变化，希望业务方主动或被动发现，有一些告警然后修改，避免数据产生一些混乱。

2.4 数据入湖痛点-数据质量管理

第三个痛点是信息入湖时会有数据质量管理上的一些痛点。例如 binlog 采集一些数据时，例如 Mysql 数据类型会比较多，需要先正确转换为 CDC 类型，再通过 CDC 管道写入到 iceberg 或 Paimon 湖的表格式中。转换过程中，这种类型例如日期时间或比较特殊的空间 GEO 类型中转到湖仓中如何做呢？做的时候会对这部分做默认的调整。例如日期，之前使用 canal 进行采集，所以一些默认值或者业务方都是基于 canal 的逻辑进行开发。所以在这部分对默认值进行改变，不能使用 DBA 采集回的一些默认值。数据质量在上线 CDC 之前会做一些数据抽样对比和整体对数，利用一些数学公式做一些交叉对比验证，对比 canal 和 CDC 整个数据链差异，确保业务方接受。没有太大误差后再投入生产 CDC。

2.5 数据入湖痛点-采集稳定性

在采集稳定性方面，存在以下问题：

（1）Schema Evolution 变更：在数据采集过程中，当 DBA 业务方需要修改一些字段时，会使用在线变更工具 GH-OST 或 PG-OST 进行变更。这些变更会导致 Schema Evolution 的变化。变更过程首先生成一个镜像表，将真实环境的数据导入镜像表，然后触发 Schema Evolution。这可能会导致某些 Schema Evolution 无法被 CDC 识别，最终导致作业的 Fellover。最新版本已经有一个 Pr 对此进行了支持。

（2）数据洪峰 GC：在采集过程中，如果遇到延迟，需要对数据进行追改，这会导致整体压力增大。因此，在 Binlog 采集的多线程处理时，需要进行调研以提升解析速度。

（3）采集告警：采集过程中会涉及 DDL 和 DML 的 Schema Change，需要对其进行预警，以关联下游所有消费的 Flink 作业，从而避免 DDL change 对业务方造成影响。

2.6 基于 Amoro+CDC 湖仓融合一体架构

基于上述提到的一些痛点，我们希望可以用一些数据湖的能力解决刚才的挑战。下面基于当前开源软件 Apache Amoro+CDC 打造湖仓融合一体的架构。下面的存储层主要是云上的对象存储，在海外使用 S3 较多，国内使用 OSS 较多。上层的一些计算层主要用到 Flink 和周边生态的一些子项目去完成数据入湖的工作，之后需要对湖进行管理。通过 Amoro 平台对这些湖仓作业进行管控优化，包括对这些表的小文件进行自动合并，合并过程比较智能，包括对湖仓的一些原数据进行统一管理和后续对索引进行优化的工作。上层主要是数据业务方和 AI 业务方例如用户画像以及数仓业务去使用，例如一些 AI 存储现在也在进一步探索，是否有可能将一些 MMA 语义和分析放到数据湖中做数据清洗或加工等操作。

上述提到 Amoro 产品，下面介绍这款产品如何与 Flink CDC 提供入湖的体验。

03.Amoro 入湖优化

3.1 湖仓管理系统 Apache Amoro

首先介绍 Apache Amoro，是去年 2 月份进入 Apache。Apache Amoro 可以管理 open formats，例如 Paimon、hudi 都可以管理，进行一些托管。在上面有对应的一些 catalog 托管在 iceberg 上。可以使用 iceberg 社区的一些协议例如 rice catalog 作为原数据的一些托管。这里提供 Self-optimizing service，基于 Spark 和 Flink 提交优化作业，与湖仓表进行绑定，自动扫描湖仓表，触发一些优化操作。

3.2 Amoro 特点

Amoro 如何做数据新鲜度的平衡呢？利用一些管理功能和自优化机制为用户提供解决三方悖论问题。从传统的数据新鲜度三角做平衡，通过 Flink 中的水印概念解决该挑战，通过自动化优化方式评估该表所需要的新鲜度，在读写上达到最佳平衡状态。

3.3 CDC 数据入湖困难点

上述提到在数据入湖上的困惑，在 Paimon 方面比较多。下面会提到 iceberg。目前 Amoro 对 iceberg 支持较多，后续会加强对 Paimon 的支持。目前 Paimon 已经支持了 catalog，但是自优化的工作还在推进。 Paimon 是基于 LSM-tree 做合并，在 iceberg 中分为 v1、v2table，对应 Paimon 中的特点 Pkatable 和 Nonpkatable 的 Attendtable。在使用时会遇到数据时效性的问题或读放大和写放大的问题。

3.4 CDC 数据入湖优化

如何解决这些问题呢？在 Amoro 中通过检测这张表，CDC 再持续写到 v2table 更新表时会产生一些小文件，在 iceberg 中分别是 pos_delete 和 eq_delete。可以看到 AMS 系统会持续监控这些表的状况，拆分成一些优化任务的 task，放在 AMS 中。AMS 下发到 Spark 或 Flink 的优化任务中。优化任务可以提交到 yaml 或 k8s 集群进行优化操作，从而达到表的读写平衡以及优化资源的隔离。

3.5 文件碎片优化过程(GC)

下面讲解文件碎片的优化过程。写入 Flink CDC 时有 Insert、Update、Delete 等操作，产生的一些小文件也会有一些标记会产生 eq-delete 和 pos-delete 的文件。优化器从 java GC 上收到启发，类似做一些并发标记的状态，分为三个阶段。第一个阶段是 Minor 阶段，做一些碎片小文件的扫描，扫描后对这些碎片文件进行消除操作，消除完之后在 Major 阶段对小文件进行进一步的去碎片化，最后达到整体数据合并可用的状态，最终我们的期望是 Data File 文件将所有 Pos-Delete 和 Eq-Delete 消除掉。

3.6 数据优化阶段

下面是三个阶段对应的一些操作。Amoro 提供了一些参数例如 CDC 在入湖时，保留多少小文件、数据新鲜度平衡时、在什么时间点做 Minor/Major 操作或 Full 操作。Full 操作相当于进行一次大的合并。都可以通过一些参数进行控制，避免在生产高峰做 Full Compassion 操作使表有性能问题。这也是刚才提到的 Paimon 合并的流程，它还是一个黑盒过程，我们希望通过该技术暴露出这部分动作，隔离出合并资源，让流作业在 CDC 入湖时更加丝滑流畅。

3.7 Flink Mixed Format

Paimon 对流的能力较好，在 iceberg 上缺乏对流的支持。Amoro 社区提供了基于 iceberg 的流的更新的能力。Flink Mixed Format 格式像 Paimon 一样支持主键表，既支持了 ODS 层的 upsert，也支持增量消费 CDC 数据，像 Paimon 的 consume 一样继续构建下游表，还可以通过 Log Store 将数据实时写入到 Kafka，为下游提供毫秒级的延迟数据。

3.8 CDC 实时入湖链路

经过一些调研和改造，我们希望可以通过 CDC 实时 ETL 写入到 iceberg 或 Paimon 中，通过 Amoro 进行湖仓表的管理动作。CDC 也可以写入到外部的存储上以及消息队列上再消费入湖。后续也可能尝试使用 Amoro 与 Flash 进行一些底层的结合，在湖仓上将数据湖的小文件或优化问题查询带来更丝滑的体验。

3.9 CDC 入湖优化效果对比

下面是合并后的优化效果。现在内部正在调研，之前做的 POC 是基于 1.20 LTS、iceberg1.6 版本写入，checkpoint 设置的是 3 分钟。蓝色是优化前文件数，紫色是未优化。可以看到优化后 CDC 在持续写入湖格式 iceberg 后，随着时间推移 iceberg 原数据文件和 data file 持续膨胀，但是在引入了 Amoro 智能化合并和湖仓的管理系统后可以更好的控制文件，可以解决一些小文件和数据索引和整体规整的一些工作。

04.未来规划

最后介绍后续的一些规划。在 CDC 社区或者 Amoro 社区将积极使用 Paimon。在之前文章中讲解了货拉拉 Lalamove 对 Paimon 的思考，可以查看之前的文章。未来希望将湖仓自优化接入到 Amoro 中，实现全自动的优化流程，将该部分暴露给用户，可以让用户看到当前的一些优化动作，即使做一些资源隔离。第二部分是 Amoro 的入湖生态。希望在 Amoro 产品上直接通过 CDC yaml 方式，用户配置完就提交作业完成整个入湖的操作。还有社区正在做的 Mixed Format 基于 iceberg 实时入湖的方案，同时 CDC 社区会支持 Pipeline 写入 iceberg sinkv2 的操作，例如稳定性的一些建设支持 FLink1.20-LTS 版本，以及与其它社区打造开源的入湖平台化方案。我们希望打造完整的入湖体验+入湖产品+自动湖仓优化的产品功能，未来社区将继续持续开展更多的合作。

更多内容

活动推荐

阿里云基于 Apache Flink 构建的企业级产品-实时计算 Flink 版现开启活动：新用户复制点击下方链接或者扫描二维码即可 0 元免费试用 Flink + Paimon实时计算 Flink 版（3000CU*小时，3 个月内）了解活动详情：https://free.aliyun.com/?utm_content=g_1000395379&productCode=sc