簡介：?介紹了 37 手游為何選擇 Flink 作為計算引擎，并如何基于 Flink CDC + Hudi 構建新的湖倉一體方案。

本文作者是 37 手游大數據開發徐潤柏，介紹了 37 手游為何選擇 Flink 作為計算引擎，并如何基于 Flink CDC + Hudi 構建新的湖倉一體方案，主要內容包括：

Flink CDC 基本知識介紹

Hudi 基本知識介紹

37 手游的業務痛點和技術方案選型

37 手游湖倉一體介紹

Flink CDC + Hudi 實踐

總結

GitHub 地址
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 詳解核心改進

二、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 數據做比對得到：

總數一致；

按天分組統計數量一致；

按天分組統計金額一致。

六、總結

湖倉一體以及流批一體架構對比傳統數倉架構主要有以下幾點好處：

• 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 的一些設計 · 語雀

原文鏈接
本文為阿里云原創內容，未經允許不得轉載。

創作挑戰賽新人創作獎勵來咯，堅持創作打卡瓜分現金大獎

總結

以上是生活随笔為你收集整理的37 手游基于 Flink CDC + Hudi 湖仓一体方案实践的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。