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

桃扇骨 2022-08-28 12:43 238阅读 0赞

**简介：** 介绍了 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.  总结

GitHub 地址  
[GitHub - apache/flink: Apache Flink][GitHub - apache_flink_ Apache Flink]  
欢迎大家给 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 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 数据的分析查询能力。

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

![img][]

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

![img][img 1]

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 让历史数据缓慢入湖。

![img][img 2]

之前 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]章节；
 *  创建 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][MySQL CDC Connector _ Flink CDC 2.0.0 documentation]

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

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

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

[GitHub - apache_flink_ Apache Flink]: https://github.com/apache/flink
[Flink CDC Connectors]: https://ververica.github.io/flink-cdc-connectors/master/index.html
[Flink CDC 2.0]: https://flink-learning.org.cn/article/detail/3ebe9f20774991c4d5eeb75a141d9e1e
[Apache Hudi]: https://hudi.apache.org/docs/overview/
[img]: /images/20220828/55538f54651a4dda8df315338e0d56e2.png
[img 1]: /images/20220828/d8a98c7450ff444785494942a5981d9b.png
[img 2]: /images/20220828/198c1fb4cc684214892334cf42295b95.png
[Hive]: https://www.yuque.com/docs/share/01c98494-a980-414c-9c45-152023bf3c17?#IsoNU
[MySQL CDC Connector _ Flink CDC 2.0.0 documentation]: https://ververica.github.io/flink-cdc-connectors/master/content/connectors/mysql-cdc.html
[HUDI FLINK _ _]: https://www.yuque.com/docs/share/01c98494-a980-414c-9c45-152023bf3c17?#
[Hudi _ _]: https://www.yuque.com/docs/share/5d1c383d-c3fc-483a-ad7e-d8181d6295cd?#
[Link 1]: http://click.aliyun.com/m/1000297487/