Flink 实践 | 37 手游基于 Flink CDC + Hudi 湖仓一体方案实践-技术圈

▼ 关注「Flink 中文社区」，获取更多技术干货 ▼

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

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

Tips：点击「阅读原文」即可查看更多技术文章～

GitHub 地址

欢迎大家给 Flink 点赞送 star~

一、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 Connectors：
https://ververica.github.io/flink-cdc-connectors/master/index.html

▼ Flink CDC 2.0 更多详情 ▼

Flink CDC 2.0 正式发布，详解核心改进

二、Hudi

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

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

Apache Hudi：
https://hudi.apache.org/docs/overview/

三、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'