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

本文涉及的产品
实时计算 Flink 版,1000CU*H 3个月
简介: 介绍了 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 地址
https://github.com/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 详解核心改进

二、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

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

之前 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

[2] Hudi Flink 答疑解惑:https://www.yuque.com/docs/share/01c98494-a980-414c-9c45-152023bf3c17?#

[3] Hudi 的一些设计:https://www.yuque.com/docs/share/5d1c383d-c3fc-483a-ad7e-d8181d6295cd?#


热点推荐

Flink Forward Asia 2021 正式启动!议题火热征集中!

30 万奖金等你来!第三届 Apache Flink 极客挑战赛暨 AAIG CUP 报名开始


更多 Flink 相关技术问题,可扫码加入社区钉钉交流群
第一时间获取最新技术文章和社区动态,请关注公众号~

image.png

活动推荐

阿里云基于 Apache Flink 构建的企业级产品-实时计算Flink版现开启活动:
99 元试用 实时计算Flink版(包年包月、10CU)即有机会获得 Flink 独家定制T恤;另包 3 个月及以上还有 85 折优惠!
了解活动详情:https://www.aliyun.com/product/bigdata/sc

image.png

相关实践学习
基于Hologres+Flink搭建GitHub实时数据大屏
通过使用Flink、Hologres构建实时数仓,并通过Hologres对接BI分析工具(以DataV为例),实现海量数据实时分析.
实时计算 Flink 实战课程
如何使用实时计算 Flink 搞定数据处理难题?实时计算 Flink 极客训练营产品、技术专家齐上阵,从开源 Flink功能介绍到实时计算 Flink 优势详解,现场实操,5天即可上手! 欢迎开通实时计算 Flink 版: https://cn.aliyun.com/product/bigdata/sc Flink Forward Asia 介绍: Flink Forward 是由 Apache 官方授权,Apache Flink Community China 支持的会议,通过参会不仅可以了解到 Flink 社区的最新动态和发展计划,还可以了解到国内外一线大厂围绕 Flink 生态的生产实践经验,是 Flink 开发者和使用者不可错过的盛会。 去年经过品牌升级后的 Flink Forward Asia 吸引了超过2000人线下参与,一举成为国内最大的 Apache 顶级项目会议。结合2020年的特殊情况,Flink Forward Asia 2020 将在12月26日以线上峰会的形式与大家见面。
相关文章
|
1月前
|
数据采集 SQL 搜索推荐
大数据之路:阿里巴巴大数据实践——OneData数据中台体系
OneData是阿里巴巴内部实现数据整合与管理的方法体系与工具,旨在解决指标混乱、数据孤岛等问题。通过规范定义、模型设计与工具平台三层架构,实现数据标准化与高效开发,提升数据质量与应用效率。
大数据之路:阿里巴巴大数据实践——OneData数据中台体系
|
2月前
|
分布式计算 监控 大数据
大数据之路:阿里巴巴大数据实践——离线数据开发
该平台提供一站式大数据开发与治理服务,涵盖数据存储计算、任务调度、质量监控及安全管控。基于MaxCompute实现海量数据处理,结合D2与DataWorks进行任务开发与运维,通过SQLSCAN与DQC保障代码质量与数据准确性。任务调度系统支持定时、周期、手动运行等多种模式,确保高效稳定的数据生产流程。
大数据之路:阿里巴巴大数据实践——离线数据开发
|
2月前
|
数据采集 存储 大数据
大数据之路:阿里巴巴大数据实践——日志采集与数据同步
本资料全面介绍大数据处理技术架构,涵盖数据采集、同步、计算与服务全流程。内容包括Web/App端日志采集方案、数据同步工具DataX与TimeTunnel、离线与实时数仓架构、OneData方法论及元数据管理等核心内容,适用于构建企业级数据平台体系。
|
2月前
|
数据采集 分布式计算 DataWorks
ODPS在某公共数据项目上的实践
本项目基于公共数据定义及ODPS与DataWorks技术,构建一体化智能化数据平台,涵盖数据目录、归集、治理、共享与开放六大目标。通过十大子系统实现全流程管理,强化数据安全与流通,提升业务效率与决策能力,助力数字化改革。
84 4
|
2月前
|
分布式计算 DataWorks 数据处理
在数据浪潮中前行:记录一次我与ODPS的实践、思考与展望
本文详细介绍了在 AI 时代背景下,如何利用阿里云 ODPS 平台(尤其是 MaxCompute)进行分布式多模态数据处理的实践过程。内容涵盖技术架构解析、完整操作流程、实际部署步骤以及未来发展方向,同时结合 CSDN 博文深入探讨了多模态数据处理的技术挑战与创新路径,为企业提供高效、低成本的大规模数据处理方案。
182 3
|
1月前
|
存储 SQL 分布式计算
大数据之路:阿里巴巴大数据实践——元数据与计算管理
本内容系统讲解了大数据体系中的元数据管理与计算优化。元数据部分涵盖技术、业务与管理元数据的分类及平台工具,并介绍血缘捕获、智能推荐与冷热分级等技术创新。元数据应用于数据标签、门户管理与建模分析。计算管理方面,深入探讨资源调度失衡、数据倾斜、小文件及长尾任务等问题,提出HBO与CBO优化策略及任务治理方案,全面提升资源利用率与任务执行效率。
|
2月前
|
机器学习/深度学习 存储 分布式计算
ODPS驱动电商仓储革命:动态需求预测系统的落地实践
本方案基于ODPS构建“预测-仿真-决策”闭环系统,解决传统仓储中滞销积压与爆款缺货问题。通过动态特征工程、时空融合模型与库存仿真引擎,实现库存周转天数下降42%,缺货率下降65%,年损减少5000万以上,显著提升运营效率与GMV。
190 1
|
6天前
|
存储 JSON 数据处理
Flink基于Paimon的实时湖仓解决方案的演进
本文源自Apache CommunityOverCode Asia 2025,阿里云专家苏轩楠分享Flink与Paimon构建实时湖仓的演进实践。深度解析Variant数据类型、Lookup Join优化等关键技术,提升半结构化数据处理效率与系统可扩展性,推动实时湖仓在生产环境的高效落地。
Flink基于Paimon的实时湖仓解决方案的演进
|
1月前
|
SQL 存储 运维
Apache Doris 在菜鸟的大规模湖仓业务场景落地实践
本文介绍了 Apache Doris 在菜鸟的大规模落地的实践经验,菜鸟为什么选择 Doris,以及 Doris 如何在菜鸟从 0 开始,一步步的验证、落地,到如今上万核的规模,服务于各个业务线,Doris 已然成为菜鸟 OLAP 数据分析的最优选型。
139 2
Apache Doris 在菜鸟的大规模湖仓业务场景落地实践
|
15天前
|
存储 人工智能 监控
淘宝闪购基于Flink&Paimon的Lakehouse生产实践:从实时数仓到湖仓一体化的演进之路
本文整理自淘宝闪购(饿了么)大数据架构师王沛斌在 Flink Forward Asia 2025 上海站的分享,深度解析其基于 Apache Flink 与 Paimon 的 Lakehouse 架构演进与落地实践,涵盖实时数仓发展、技术选型、平台建设及未来展望。
147 0
淘宝闪购基于Flink&Paimon的Lakehouse生产实践:从实时数仓到湖仓一体化的演进之路

热门文章

最新文章

相关产品

  • 实时计算 Flink版