37 手游基于 Flink CDC + Hudi 湖仓一体方案实践

​简介： 介绍了 37 手游为何选择 Flink 作为计算引擎，并如何基于 Flink CDC + Hudi 构建新的湖仓一体方案。

本文作者是 37 手游大数据开发徐润柏，介绍了 37 手游为何选择 Flink 作为计算引擎，并如何基于 Flink CDC + Hudi 构建新的湖仓一体方案，主要内容包括：

1. Flink CDC 基本知识介绍
2. Hudi 基本知识介绍
3. 37 手游的业务痛点和技术方案选型
4. 37 手游湖仓一体介绍
5. Flink CDC + Hudi 实践
6. 总结

一、Flink-CDC 2.0

Flink CDC Connectors 是 Apache Flink 的一个 source 端的连接器，目前 2.0 版本支持从 MySQL 以及 Postgres 两种数据源中获取数据，2.1 版本社区确定会支持 Oracle，MongoDB 数据源。

Fink CDC 2.0 的核心 feature，主要表现为实现了以下三个非常重要的功能：

• 全程无锁，不会对数据库产生需要加锁所带来的风险；
• 多并行度，全量数据的读取阶段支持水平扩展，使亿级别的大表可以通过加大并行度来加快读取速度；
• 断点续传，全量阶段支持 checkpoint，即使任务因某种原因退出了，也可通过保存的 checkpoint 对任务进行恢复实现数据的断点续传。

Flink CDC 2.0 详解核心改进

二、Hudi

Apache Hudi 目前被业内描述为围绕数据库内核构建的流式数据湖平台 (Streaming Data Lake Platform)。

由于 Hudi 拥有良好的 Upsert 能力，并且 0.10 Master 对 Flink 版本支持至 1.13.x，因此我们选择通过 Flink + Hudi 的方式为 37 手游的业务场景提供分钟级 Upsert 数据的分析查询能力。

三、37 手游的业务痛点和技术方案选型

​

1. 旧架构与业务痛点

1.1 数据实时性不够

• 日志类数据通过 sqoop 每 30min 同步前 60min 数据到 Hive；
• 数据库类数据通过 sqoop 每 60min 同步当天全量数据到 Hive；
• 数据库类数据通过 sqoop 每天同步前 60 天数据到 Hive。

1.2 业务代码逻辑复杂且难维护

• 目前 37 手游还有很多的业务开发沿用 MySQL + PHP 的开发模式，代码逻辑复杂且很难维护；
• 相同的代码逻辑，往往流处理需要开发一份代码，批处理则需要另开发一份代码，不能复用。

1.3 频繁重刷历史数据

• 频繁地重刷历史数据来保证数据一致。

1.4 Schema 变更频繁

• 由于业务需求，经常需要添加表字段。

1.5 Hive 版本低

• 目前 Hive 使用版本为 1.x 版本，并且升级版本比较困难；
• 不支持 Upsert；
• 不支持行级别的 delete。

由于 37 手游的业务场景，数据 upsert、delete 是个很常见的需求。所以基于 Hive 数仓的架构对业务需求的满足度不够。

2. 技术选型

在同步工具的选型上考虑过 Canal 和 Maxwell。但 Canal 只适合增量数据的同步并且需要部署，维护起来相对较重。而 Maxwell 虽然比较轻量，但与 Canal 一样需要配合 Kafka 等消息队列使用。对比之下，Flink CDC 可以通过配置 Flink connector 的方式基于 Flink-SQL 进行使用，十分轻巧，并且完美契合基于 Flink-SQL 的流批一体架构。

在存储引擎的选型上，目前最热门的数据湖产品当属：Apache Hudi，Apache Iceberg 和 DeltaLake，这些在我们的场景下各有优劣。最终，基于 Hudi 对上下游生态的开放、对全局索引的支持、对 Flink 1.13 版本的支持，以及对 Hive 版本的兼容性 (Iceberg 不支持 Hive1.x 的版本) 等原因，选择了 Hudi 作为湖仓一体和流批一体的存储引擎。

针对上述存在的业务痛点以及选型对比，我们的最终方案为：以 Flink1.13.2 作为计算引擎，依靠 Flink 提供的流批统一的 API，基于 Flink-SQL 实现流批一体，Flink-CDC 2.0 作为 ODS 层的数据同步工具以及 Hudi-0.10 Master 作为存储引擎的湖仓一体，解决维护两套代码的业务痛点。

四、新架构与湖仓一体

37 手游的湖仓一体方案，是 37 手游流批一体架构的一部分。通过湖仓一体、流批一体，准实时场景下做到了：数据同源、同计算引擎、同存储、同计算口径。数据的时效性可以到分钟级，能很好的满足业务准实时数仓的需求。下面是架构图：

​

MySQL 数据通过 Flink CDC 进入到 Kafka。之所以数据先入 Kafka 而不是直接入 Hudi，是为了实现多个实时任务复用 MySQL 过来的数据，避免多个任务通过 Flink CDC 接 MySQL 表以及 Binlog，对 MySQL 库的性能造成影响。

通过 CDC 进入到 Kafka 的数据除了落一份到离线数据仓库的 ODS 层之外，会同时按照实时数据仓库的链路，从 ODS->DWD->DWS->OLAP 数据库，最后供报表等数据服务使用。实时数仓的每一层结果数据会准实时的落一份到离线数仓，通过这种方式做到程序一次开发、指标口径统一，数据统一。

从架构图上，可以看到有一步数据修正 (重跑历史数据) 的动作，之所以有这一步是考虑到：有可能存在由于口径调整或者前一天的实时任务计算结果错误，导致重跑历史数据的情况。

而存储在 Kafka 的数据有失效时间，不会存太久的历史数据，重跑很久的历史数据无法从 Kafka 中获取历史源数据。再者，如果把大量的历史数据再一次推到 Kafka，走实时计算的链路来修正历史数据，可能会影响当天的实时作业。所以针对重跑历史数据，会通过数据修正这一步来处理。

总体上说，37 手游的数据仓库属于 Lambda 和 Kappa 混搭的架构。流批一体数据仓库的各个数据链路有数据质量校验的流程。第二天对前一天的数据进行对账，如果前一天实时计算的数据无异常，则不需要修正数据，Kappa 架构已经足够。

五、Flink CDC 2.0 + Kafka + Hudi 0.10 实践

1. 环境准备

• Flink 1.13.2
• .../lib/hudi-flink-bundle_2.11-0.10.0-SNAPSHOT.jar (修改 Master 分支的 Hudi Flink 版本为 1.13.2 然后构建)
• .../lib/hadoop-mapreduce-client-core-2.7.3.jar (解决 Hudi ClassNotFoundException)
• ../lib/flink-sql-connector-mysql-cdc-2.0.0.jar
• ../lib/flink-format-changelog-json-2.0.0.jar
• ../lib/flink-sql-connector-kafka_2.11-1.13.2.jar

source 端 MySQL-CDC 表定义：

create table sy_payment_cdc (
  ID BIGINT,
  ...
  PRIMARY KEY(ID) NOT ENFORCED
) with(
  'connector' = 'mysql-cdc',
  'hostname' = '',
  'port' = '',
  'username' = '',
  'password' = '',
  'database-name' = '',
  'table-name' = '',
  'connect.timeout' = '60s',
  'scan.incremental.snapshot.chunk.size' = '100000',
  'server-id'='5401-5416'
);

值得注意的是：scan.incremental.snapshot.chunk.size 参数需要根据实际情况来配置，如果表数据量不大，使用默认值即可。

Sink 端 Kafka+Hudi COW 表定义：

create table sy_payment_cdc2kafka (
  ID BIGINT,
  ...
  PRIMARY KEY(ID) NOT ENFORCED
) with (
  'connector' = 'kafka',
  'topic' = '',
  'scan.startup.mode' = 'latest-offset',
  'properties.bootstrap.servers' = '',
  'properties.group.id' = '',
  'key.format' = '',
  'key.fields' = '',
  'format' = 'changelog-json'
);

create table sy_payment2Hudi (
  ID BIGINT,
  ...
  PRIMARY KEY(ID) NOT ENFORCED
)
PARTITIONED BY (YMD)
WITH (
  'connector' = 'Hudi',
  'path' = 'hdfs:///data/Hudi/m37_mpay_tj/sy_payment',
  'table.type' = 'COPY_ON_WRITE',
  'partition.default_name' = 'YMD',
  'write.insert.drop.duplicates' = 'true',
  'write.bulk_insert.shuffle_by_partition' = 'false',
  'write.bulk_insert.sort_by_partition' = 'false',
  'write.precombine.field' = 'MTIME',
  'write.tasks' = '16',
  'write.bucket_assign.tasks' = '16',
  'write.task.max.size' = '',
  'write.merge.max_memory' = ''
);

针对历史数据入 Hudi，可以选择离线 bulk_insert 的方式入湖，再通过 Load Index Bootstrap 加载数据后接回增量数据。bulk_insert 方式入湖数据的唯一性依靠源端的数据本身，在接回增量数据时也需要做到保证数据不丢失。

这里我们选择更为简单的调整任务资源的方式将历史数据入湖。依靠 Flink 的 checkpoint 机制，不管是 CDC 2.0 入 Kafka 期间还是 Kafka 入 Hudi 期间，都可以通过指定 checkpoint 的方式对任务进行重启并且数据不会丢失。

我们可以在配置 CDC 2.0 入 Kafka，Kafka 入 Hudi 任务时调大内存并配置多个并行度，加快历史数据入湖，等到所有历史数据入湖后，再相应的调小入湖任务的内存配置并且将 CDC 入 Kafka 的并行度设置为 1，因为增量阶段 CDC 是单并行度，然后指定 checkpoint 重启任务。

按照上面表定义的参数配置，配置 16 个并行度，Flink TaskManager 内存大小为 50G 的情况下，单表 15 亿历史数据入至 Hudi COW 表实际用时 10 小时，单表 9 亿数据入至 Hudi COW 表实际用时 6 小时。当然这个耗时很大一部分是 COW 写放大的特性，在大数据量的 upsert 模式下耗时较多。

目前我们的集群由 200 多台机器组成，在线的流计算任务总数有 200 多，总数据量接近 2PB。

如果集群资源很有限的情况下，可以根据实际情况调整 Hudi 表以及 Flink 任务的内存配置，还可以通过配置 Hudi 的限流参数 write.rate.limit 让历史数据缓慢入湖。

​

之前 Flink CDC 1.x 版本由于全量 snapshot 阶段单并行度读取的原因，当时亿级以上的表在全量 snapshot 读取阶段就需要耗费很长时间，并且 checkpoint 会失败无法保证数据的断点续传。

所以当时入 Hudi 是采用先启动一个 CDC 1.x 的程序将此刻开始的增量数据写入 Kafka，之后再启动另外一个 sqoop 程序拉取当前的所有数据至 Hive 后，通过 Flink 读取 Hive 的数据写 Hudi，最后再把 Kafka 的增量数据从头消费接回 Hudi。由于 Kafka 与 Hive 的数据存在交集，因此数据不会丢失，加上 Hudi 的 upsert 能力保证了数据唯一。

但是，这种方式的链路太长操作困难，如今通过 CDC 2.0 在全量 snapshot 阶段支持多并行度以及 checkpoint 的能力，确实大大降低了架构的复杂度。

2. 数据比对

• 由于生产环境用的是 Hive1.x，Hudi 对于 1.x 还不支持数据同步，所以通过创建 Hive 外部表的方式进行查询，如果是 Hive2.x 以上版本，可参考 Hive 同步章节；
• 创建 Hive 外部表 + 预创建分区；
• auxlib 文件夹添加 Hudi-hadoop-mr-bundle-0.10.0-SNAPSHOT.jar。

CREATE EXTERNAL TABLE m37_mpay_tj.`ods_sy_payment_f_d_b_ext`(
  `_hoodie_commit_time` string,
  `_hoodie_commit_seqno` string,
  `_hoodie_record_key` string,
  `_hoodie_partition_path` string,
  `_hoodie_file_name` string,
  `ID` bigint,
  ...
  )
PARTITIONED BY (
  `dt` string)
ROW FORMAT SERDE
  'org.apache.hadoop.Hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT
  'org.apache.Hudi.hadoop.HoodieParquetInputFormat'
OUTPUTFORMAT
  'org.apache.hadoop.Hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
  'hdfs:///data/Hudi/m37_mpay_tj/sy_payment'

最终查询 Hudi 数据 (Hive 外部表的形式) 与原来 sqoop 同步的 Hive 数据做比对得到：

1. 总数一致；
2. 按天分组统计数量一致；
3. 按天分组统计金额一致。

六、总结

湖仓一体以及流批一体架构对比传统数仓架构主要有以下几点好处：

• Hudi 提供了 Upsert 能力，解决频繁 Upsert/Delete 的痛点；
• 提供分钟级的数据，比传统数仓有更高的时效性；
• 基于 Flink-SQL 实现了流批一体，代码维护成本低；
• 数据同源、同计算引擎、同存储、同计算口径；
• 选用 Flink CDC 作为数据同步工具，省掉 sqoop 的维护成本。

最后针对频繁增加表字段的痛点需求，并且希望后续同步下游系统的时候能够自动加入这个字段，目前还没有完美的解决方案，希望 Flink CDC 社区能在后续的版本提供 Schema Evolution 的支持。

Reference

[1] MySQL CDC 文档： MySQL CDC Connector — Flink CDC 2.0.0 documentation

[2] Hudi Flink 答疑解惑：HUDI FLINK 答疑解惑 · 语雀

[3] Hudi 的一些设计：Hudi 的一些设计 · 语雀

原文链接
本文为阿里云原创内容，未经允许不得转载。

​