基于Flink CDC打通業(yè)務(wù)數(shù)據(jù)實(shí)時(shí)入湖

作者：數(shù)據(jù)社 2021-06-04 07:24:14

在構(gòu)建實(shí)時(shí)數(shù)倉(cāng)的過(guò)程中，如何快速、正確的同步業(yè)務(wù)數(shù)據(jù)是最先面臨的問(wèn)題，本文主要討論一下如何使用實(shí)時(shí)處理引擎Flink和數(shù)據(jù)湖Apache Iceberg兩種技術(shù)，來(lái)解決業(yè)務(wù)數(shù)據(jù)實(shí)時(shí)入湖相關(guān)的問(wèn)題。

[[403582]]

大家好，我是一哥，今天分享一篇數(shù)據(jù)實(shí)時(shí)入湖的干貨文章。

01 Flink CDC介紹

CDC全稱(chēng)是Change Data Capture，捕獲變更數(shù)據(jù)，是一個(gè)比較廣泛的概念，只要是能夠捕獲所有數(shù)據(jù)的變化，比如數(shù)據(jù)庫(kù)捕獲完整的變更日志記錄增、刪、改等，都可以稱(chēng)為CDC。該功能被廣泛應(yīng)用于數(shù)據(jù)同步、更新緩存、微服務(wù)間同步數(shù)據(jù)等場(chǎng)景，本文主要介紹基于Flink CDC在數(shù)據(jù)實(shí)時(shí)同步場(chǎng)景下的應(yīng)用。

Flink在1.11版本開(kāi)始引入了Flink CDC功能，并且同時(shí)支持Table & SQL兩種形式。Flink SQL CDC是以SQL的形式編寫(xiě)實(shí)時(shí)任務(wù)，并對(duì)CDC數(shù)據(jù)進(jìn)行實(shí)時(shí)解析同步。相比于傳統(tǒng)的數(shù)據(jù)同步方案，該方案在實(shí)時(shí)性、易用性等方面有了極大的改善。下圖是基于Flink SQL CDC的數(shù)據(jù)同步方案的示意圖。

Oracle的變更日志的采集有多種方案，這里采用的Debezium實(shí)時(shí)同步工具作為示例，該工具能夠解析Oracle的changlog數(shù)據(jù)，并實(shí)時(shí)同步數(shù)據(jù)到下游Kafka。Flink SQL通過(guò)創(chuàng)建Kafka映射表并指定 format格式為debezium-json，然后通過(guò)Flink進(jìn)行解析后直接插入到其他外部數(shù)據(jù)存儲(chǔ)系統(tǒng)，例如圖中外部數(shù)據(jù)源以Apache Iceberg為例。

下面詳細(xì)解析一下數(shù)據(jù)同步過(guò)程。首先了解一下Debezium抽取的Oracle的change log的格式，以u(píng)pdate為例，變更日志上記錄了更新之前的數(shù)據(jù)和更新以后的數(shù)據(jù)，在Kafka下游的Flink接受到這樣的數(shù)據(jù)以后，一條update操作記錄就轉(zhuǎn)變?yōu)榱讼萪elete、后insert兩條記錄。日志格式如下所示，該update操作的內(nèi)容的name字段從tom更新為了jerry。

{ 
  "before": {          --更新之前的數(shù)據(jù) 
    "id": 001, 
    "name": "tom" 
  }, 
  "after": {           --更新之后的數(shù)據(jù) 
    "id": 001, 
    "name": "jerry" 
  }, 
  "source": {...}, 
  "op": "u", 
  "ts_ms": 1589362330904, 
  "transaction": null 
}

其次再來(lái)看一下Flink SQL內(nèi)部是如何處理update記錄的。Flink在1.11版本支持了完整的changelog機(jī)制，對(duì)于每條數(shù)據(jù)本身只要是攜帶了相應(yīng)增、刪、改的標(biāo)志，F(xiàn)link就能識(shí)別這些數(shù)據(jù)，并對(duì)結(jié)果表做出相應(yīng)的增、刪、改的動(dòng)作，如下圖所示changlog數(shù)據(jù)流經(jīng)過(guò)Flink解析，同步到下游Sink Database。

通過(guò)以上分析，基于Flink SQL CDC的數(shù)據(jù)同步有如下優(yōu)點(diǎn)：

業(yè)務(wù)解耦：無(wú)需入侵業(yè)務(wù)，和業(yè)務(wù)完全解耦，也就是業(yè)務(wù)端無(wú)感知數(shù)據(jù)同步的存在。
性能消耗：業(yè)務(wù)數(shù)據(jù)庫(kù)性能消耗小，數(shù)據(jù)同步延遲低。
同步易用：使用SQL方式執(zhí)行CDC同步任務(wù)，極大的降低使用維護(hù)門(mén)檻。
數(shù)據(jù)完整：完整的數(shù)據(jù)庫(kù)變更記錄，不會(huì)丟失任何記錄，F(xiàn)link 自身支持 Exactly Once。

02 Apache Iceberg介紹

通常認(rèn)為數(shù)據(jù)湖是一種支持存儲(chǔ)多種原始數(shù)據(jù)格式、多種計(jì)算引擎、高效的元數(shù)據(jù)統(tǒng)一管理和海量統(tǒng)一數(shù)據(jù)存儲(chǔ)。其中以Apache Iceberg為代表的表格式和Flink計(jì)算引擎組成的數(shù)據(jù)湖解決方案尤為亮眼。Flink社區(qū)方面也主動(dòng)擁抱數(shù)據(jù)湖技術(shù)，當(dāng)前Flink和Iceberg在數(shù)據(jù)入湖方面的集成度最高。

那么Apache Iceberg是什么呢?引用官網(wǎng)的定義是：Apache Iceberg is an open table format for huge analytic datasets。也就是Apache Iceberg是一個(gè)大規(guī)模數(shù)據(jù)分析的開(kāi)放表格式。

Iceberg將數(shù)據(jù)分為元數(shù)據(jù)管理層和數(shù)據(jù)存儲(chǔ)層。首先了解一下Iceberg在文件系統(tǒng)中的布局，第一部分是數(shù)據(jù)文件data files，用于存儲(chǔ)具體業(yè)務(wù)數(shù)據(jù)，如下圖中的data files文件。第二部分是表元數(shù)據(jù)文件(Metadata 文件)，包含Snapshot文件、Manifest文件等。Snapshot表示當(dāng)前操作的一個(gè)快照，每次commit都會(huì)生成一個(gè)快照，一個(gè)快照中包含多個(gè)Manifest，每個(gè)Manifest中記錄了當(dāng)前操作生成數(shù)據(jù)所對(duì)應(yīng)的文件地址，也就是data files的地址。基于snapshot的管理方式，iceberg能夠進(jìn)行time travel(歷史版本讀取以及增量讀取)。Iceberg文件系統(tǒng)設(shè)計(jì)特點(diǎn)如下圖所示：

Iceberg的表格式設(shè)計(jì)具有如下特點(diǎn)：

ACID：不會(huì)讀到不完整的commit數(shù)據(jù)，基于樂(lè)觀鎖實(shí)現(xiàn)，支持并發(fā)commit，支持Row-level delete，支持upsert操作。
增量快照：Commit后的數(shù)據(jù)即可見(jiàn)，在Flink實(shí)時(shí)入湖場(chǎng)景下，數(shù)據(jù)可見(jiàn)根據(jù)checkpoint的時(shí)間間隔來(lái)確定的，增量形式也可回溯歷史快照。
開(kāi)放的表格式：對(duì)于一個(gè)真正的開(kāi)放表格式，支持多種數(shù)據(jù)存儲(chǔ)格式，如：parquet、orc、avro等，支持多種計(jì)算引擎，如：Spark、Flink、Hive、Trino/Presto。
流批接口支持：支持流式寫(xiě)入、批量寫(xiě)入，支持流式讀取、批量讀取。下文的測(cè)試中，主要測(cè)試了流式寫(xiě)入和批量讀取的功能。

03 Flink CDC打通數(shù)據(jù)實(shí)時(shí)導(dǎo)入Iceberg實(shí)踐

當(dāng)前使用Flink最新版本1.12，支持CDC功能和更好的流批一體。Apache Iceberg最新版本0.11已經(jīng)支持Flink API方式upsert，如果使用編寫(xiě)框架代碼的方式使用該功能，無(wú)異于鏡花水月，可望而不可及。本著SQL就是生產(chǎn)力的初衷，該測(cè)試使用最新Iceberg的master分支代碼編譯嘗鮮，并對(duì)源碼稍做修改，達(dá)到支持使用Flink SQL方式upsert。

先來(lái)了解一下什么是Row-Level Delete?該功能是指根據(jù)一個(gè)條件從一個(gè)數(shù)據(jù)集里面刪除指定行。那么為什么這個(gè)功能那么重要呢?眾所周知，大數(shù)據(jù)中的行級(jí)刪除不同于傳統(tǒng)數(shù)據(jù)庫(kù)的更新和刪除功能，在基于HDFS架構(gòu)的文件系統(tǒng)上數(shù)據(jù)存儲(chǔ)只支持?jǐn)?shù)據(jù)的追加，為了在該構(gòu)架下支持更新刪除功能，刪除操作演變成了一種標(biāo)記刪除，更新操作則是轉(zhuǎn)變?yōu)橄葮?biāo)記刪除、后插入一條新數(shù)據(jù)。具體實(shí)現(xiàn)方式可以分為Copy on Write(COW)模式和Merge on Read(MOR)模式，其中Copy on Write模式可以保證下游的數(shù)據(jù)讀具有最大的性能，而Merge on Read模式保證上游數(shù)據(jù)插入、更新、和刪除的性能，減少傳統(tǒng)Copy on Write模式下寫(xiě)放大問(wèn)題。

在Apache Iceberg中目前實(shí)現(xiàn)的是基于Merge on Read模式實(shí)現(xiàn)的Row-Level Delete。在 Iceberg中MOR相關(guān)的功能是在Iceberg Table Spec Version 2: Row-level Deletes中進(jìn)行實(shí)現(xiàn)的，V1是沒(méi)有相關(guān)實(shí)現(xiàn)的。雖然當(dāng)前Apache Iceberg 0.11版本不支持Flink SQL方式進(jìn)行Row-Level Delete，但為了方便測(cè)試，通過(guò)對(duì)源碼的修改支持Flink SQL方式。在不遠(yuǎn)的未來(lái)，Apache Iceberg 0.12版本將會(huì)對(duì)Row-Level Delete進(jìn)行性能和穩(wěn)定性的加強(qiáng)。

Flink SQL CDC和Apache Iceberg的架構(gòu)設(shè)計(jì)和整合如何巧妙，不能局限于紙上談兵，下面就實(shí)際操作一下，體驗(yàn)其功能的強(qiáng)大和帶來(lái)的便捷。并且順便體驗(yàn)一番流批一體，下面的離線(xiàn)查詢(xún)和實(shí)時(shí)upsert入湖等均使用Flink SQL完成。

1，數(shù)據(jù)入湖環(huán)境準(zhǔn)備

以Flink SQL CDC方式將實(shí)時(shí)數(shù)據(jù)導(dǎo)入數(shù)據(jù)湖的環(huán)境準(zhǔn)備非常簡(jiǎn)單直觀，因?yàn)镕link支持流批一體功能，所以實(shí)時(shí)導(dǎo)入數(shù)據(jù)湖的數(shù)據(jù)，也可以使用Flink SQL離線(xiàn)或?qū)崟r(shí)進(jìn)行查詢(xún)。如下測(cè)試是使用Flink提供的sql-client完成的：

第一步，新建Kafka映射表，用于實(shí)時(shí)接收Topic中的changlog數(shù)據(jù)：

id STRING, 
name STRING 
) WITH ( 
'connector' = 'kafka', 
'topic' = 'topic_name', 
'properties.bootstrap.servers' = 'localhost:9092', 
'properties.group.id' = 'testGroup', 
'scan.startup.mode' = 'earliest-offset', 
'format' = 'debezium-json'

第二步，新建iceberg結(jié)果表，用于存儲(chǔ)實(shí)時(shí)入湖的數(shù)據(jù)：

CREATE TABLE iceberg_catalog.default.IcebergTable ( id STRING, name STRING );

注：

a)其中省略了配置catalog過(guò)程

b)當(dāng)前iceberg 0.11默認(rèn)創(chuàng)建表格式版本V1，通過(guò)代碼更改版本為V2，以支持upsert方式導(dǎo)入數(shù)據(jù)湖

第三步，啟動(dòng)upsert方式實(shí)時(shí)入湖的Flink任務(wù)：

SET table.dynamic-table-options.enabled=true; 
 
INSERT INTO IcebergTable /*+OPTIONS('equality-field-columns'='id')*/ SELECT * FROM KafkaTable;

注：當(dāng)前iceberg 0.11不支持Flink SQL形式upsert，通過(guò)修改源碼達(dá)到支持配置指定字段更新功能。

第四步，離線(xiàn)或者實(shí)時(shí)查詢(xún)統(tǒng)計(jì)IcebergTable表中的數(shù)據(jù)行數(shù)：

a)離線(xiàn)方式

SET execution.type=batch; 
SELECT COUNT(*) FROM IcebergTable;

b)實(shí)時(shí)方式

SET execution.type=streaming; 
 
SELECT COUNT(*) FROM IcebergTable;

2，數(shù)據(jù)入湖速度測(cè)試

數(shù)據(jù)入湖速度測(cè)試會(huì)根據(jù)環(huán)境配置、參數(shù)配置、數(shù)據(jù)格式等不同有所不同，下面是列出主要配置和測(cè)試出的數(shù)據(jù)作為參考。

a)資源配置情況

TaskManager 內(nèi)存4G，slot：1 
Checkpoint 1分鐘 
測(cè)試數(shù)據(jù)列數(shù) 10列 
測(cè)試數(shù)據(jù)行數(shù) 1000萬(wàn) 
iceberg存儲(chǔ)格式 parquet

b)測(cè)試數(shù)據(jù)情況

數(shù)據(jù)入湖分為append和upsert兩種方式。append方式只能新增數(shù)據(jù)，不能對(duì)結(jié)果數(shù)據(jù)進(jìn)行更新操作;upsert方式即能夠?qū)Y(jié)果數(shù)據(jù)更新。

append方式使用場(chǎng)景是導(dǎo)入數(shù)據(jù)之前已經(jīng)明確該表數(shù)據(jù)不需要更新，如離線(xiàn)數(shù)據(jù)導(dǎo)入數(shù)據(jù)湖的場(chǎng)景，append方式下導(dǎo)入數(shù)據(jù)速度如下：

INSERT INTO IcebergTable SELECT * FROM KafkaTable; 
 
并行度1 12.2萬(wàn)/秒 
并行度2 19.6萬(wàn)/秒 
并行度4 28.3萬(wàn)/秒

update方式使用場(chǎng)景是既有插入的數(shù)據(jù)又有對(duì)之前插入數(shù)據(jù)的更新的場(chǎng)景，如數(shù)據(jù)庫(kù)實(shí)時(shí)同步，upsert方式下導(dǎo)入數(shù)據(jù)速度，該方式需要指定在更新時(shí)以那個(gè)字段查找，類(lèi)似于update語(yǔ)句中的where條件，一般設(shè)置為表的主鍵即可，如下：

INSERT INTO IcebergTable /*+OPTIONS('equality-field-columns'='id')*/ SELECT * FROM KafkaTable; 
 
導(dǎo)入的數(shù)據(jù) 只有數(shù)據(jù)插入 只有數(shù)據(jù)更新 
并行度1 3.2萬(wàn)/秒 2.9萬(wàn)/秒 
并行度2 4.9萬(wàn)/秒 4.2萬(wàn)/秒 
并行度4 6.1萬(wàn)/秒 5.1萬(wàn)/秒

c)結(jié)論

append方式導(dǎo)入速度遠(yuǎn)大于upsert導(dǎo)入數(shù)據(jù)速度。在使用的時(shí)候，如沒(méi)有更新數(shù)據(jù)的場(chǎng)景時(shí)，則不需要upsert方式導(dǎo)入數(shù)據(jù)。

導(dǎo)入速度隨著并行度的增加而增加。

upsert方式數(shù)據(jù)的插入和更新速度相差不大，主要得益于MOR原因。

3，數(shù)據(jù)入湖任務(wù)運(yùn)維

在實(shí)際使用過(guò)程中，默認(rèn)配置下是不能夠長(zhǎng)期穩(wěn)定的運(yùn)行的，一個(gè)實(shí)時(shí)數(shù)據(jù)導(dǎo)入iceberg表的任務(wù)，需要通過(guò)至少下述四點(diǎn)進(jìn)行維護(hù)，才能使Iceberg表的入湖和查詢(xún)性能保持穩(wěn)定。

a)壓縮小文件

Flink從Kafka消費(fèi)的數(shù)據(jù)以checkpoint方式提交到Iceberg表，數(shù)據(jù)文件使用的是parquet格式，這種格式無(wú)法追加，而流式數(shù)據(jù)又不能等候太長(zhǎng)時(shí)間，所以會(huì)不斷commit提交數(shù)據(jù)產(chǎn)生小文件。目前Iceberg提供了一個(gè)批任務(wù)action來(lái)壓縮小文件，需要定期周期性調(diào)用進(jìn)行小文件的壓縮功能。示例代碼如下：

Table table = ...  
Actions.forTable(table) 
.rewriteDataFiles() 
    .targetSizeInBytes(100 * 1024 * 1024) // 100 MB 
    .execute();

b)快照過(guò)期處理

iceberg本身的架構(gòu)設(shè)計(jì)決定了，對(duì)于實(shí)時(shí)入湖場(chǎng)景，會(huì)產(chǎn)生大量的snapshot文件，快照過(guò)期策略是通過(guò)額外的定時(shí)任務(wù)周期執(zhí)行，過(guò)期snapshot文件和過(guò)期數(shù)據(jù)文件均會(huì)被刪除。如果實(shí)際使用場(chǎng)景不需要time travel功能，則可以保留較少的snapshot文件。

Table table = ...  
Actions.forTable(table) 
    .expireSnapshots() 
.expireOlderThan(System.currentTimeMillis()) 
.retainLast(5) 
    .execute();

c)清理orphan文件

orphan文件的產(chǎn)生是由于正常或者異常的數(shù)據(jù)寫(xiě)入但是未提交導(dǎo)致的，長(zhǎng)時(shí)間積累會(huì)產(chǎn)生大量脫離元數(shù)據(jù)的孤立數(shù)據(jù)文件，所以也需要類(lèi)似JVM的垃圾回收一樣，周期性清理這些文件。該功能不需要頻繁運(yùn)行，設(shè)置為3-5天運(yùn)行一次即可。

Table table = ... 
Actions.forTable(table) 
    .removeOrphanFiles() 
    .execute();

d)刪除元數(shù)據(jù)文件

每次提交snapshot均會(huì)自動(dòng)產(chǎn)生一個(gè)新的metadata文件，實(shí)時(shí)數(shù)據(jù)入庫(kù)會(huì)頻繁的產(chǎn)生大量metadata文件，需要通過(guò)如下配置達(dá)到自動(dòng)刪除metadata文件的效果。

Property	Description
write.metadata.delete-after-commit.enabled	Whether to delete old metadata files after each table commit
write.metadata.previous-versions-max	The number of old metadata files to keep

4，數(shù)據(jù)入湖問(wèn)題討論

這里主要討論數(shù)據(jù)一致性和順序性問(wèn)題。

Q1: 程序BUG或者任務(wù)重啟等導(dǎo)致數(shù)據(jù)傳輸中斷，如何保證數(shù)據(jù)的一致性呢?

Answer：數(shù)據(jù)一致保證通過(guò)兩個(gè)方面實(shí)現(xiàn)，借助Flink實(shí)現(xiàn)的exactly once語(yǔ)義和故障恢復(fù)能力，實(shí)現(xiàn)數(shù)據(jù)嚴(yán)格一致性。借助Iceberg ACID能力來(lái)隔離寫(xiě)入對(duì)分析任務(wù)的不利影響。

Q2：數(shù)據(jù)入湖否可保證全局順序性插入和更新?

Answer：不可以全局保證數(shù)據(jù)生產(chǎn)和數(shù)據(jù)消費(fèi)的順序性，但是可以保證同一條數(shù)據(jù)的插入和更新的順序性。首先數(shù)據(jù)抽取的時(shí)候是單線(xiàn)程的，然后分發(fā)到Kafka的各個(gè)partition中，此時(shí)同一個(gè)key的變更數(shù)據(jù)打入到同一個(gè)Kafka的分區(qū)里面，F(xiàn)link讀取的時(shí)候也能保證順序性消費(fèi)每個(gè)分區(qū)中的數(shù)據(jù)，進(jìn)而保證同一個(gè)key的插入和更新的順序性。

04 未來(lái)規(guī)劃

新的技術(shù)最終是要落地才能發(fā)揮其內(nèi)在價(jià)值的，針對(duì)在實(shí)踐應(yīng)用中面臨的紛繁復(fù)雜的數(shù)據(jù)，結(jié)合流計(jì)算技術(shù)Flink、表格式Iceberg，未來(lái)落地規(guī)劃主要集中在兩個(gè)方面，一是Iceberg集成到本行的實(shí)時(shí)計(jì)算平臺(tái)中，解決易用性的問(wèn)題;二是基于Iceberg，構(gòu)建準(zhǔn)實(shí)時(shí)數(shù)倉(cāng)進(jìn)行探索和落地。

1，整合Iceberg到實(shí)時(shí)計(jì)算平臺(tái)

目前，我所負(fù)責(zé)的實(shí)時(shí)計(jì)算平臺(tái)是一個(gè)基于SQL的高性能實(shí)時(shí)大數(shù)據(jù)處理平臺(tái)，該平臺(tái)徹底規(guī)避繁重的底層流計(jì)算處理邏輯、繁瑣的提交過(guò)程等，為用戶(hù)打造一個(gè)只需關(guān)注實(shí)時(shí)計(jì)算邏輯的平臺(tái)，助力企業(yè)向?qū)崟r(shí)化、智能化大數(shù)據(jù)轉(zhuǎn)型。

實(shí)時(shí)計(jì)算平臺(tái)未來(lái)將會(huì)整合Apache Iceberg數(shù)據(jù)源，用戶(hù)可以在界面配置Flink SQL任務(wù)，該任務(wù)以u(píng)psert方式實(shí)時(shí)解析changlog并導(dǎo)入到數(shù)據(jù)湖中。并增加小文件監(jiān)控、定時(shí)任務(wù)壓縮小文件、清理過(guò)期數(shù)據(jù)等功能。

2，準(zhǔn)實(shí)時(shí)數(shù)倉(cāng)探索

本文對(duì)數(shù)據(jù)實(shí)時(shí)入湖從原理和實(shí)戰(zhàn)做了比較多的闡述，在完成實(shí)時(shí)數(shù)據(jù)入湖SQL化的功能以后，入湖后的數(shù)據(jù)有哪些場(chǎng)景的使用呢?下一個(gè)目標(biāo)當(dāng)然是入湖的數(shù)據(jù)分析實(shí)時(shí)化。比較多的討論是關(guān)于實(shí)時(shí)數(shù)據(jù)湖的探索，結(jié)合所在企業(yè)數(shù)據(jù)特點(diǎn)探索適合落地的實(shí)時(shí)數(shù)據(jù)分析場(chǎng)景成為當(dāng)務(wù)之急。

隨著數(shù)據(jù)量的持續(xù)增大，和業(yè)務(wù)對(duì)時(shí)效性的嚴(yán)苛要求，基于Apache Flink和Apache Iceberg構(gòu)建準(zhǔn)實(shí)時(shí)數(shù)倉(cāng)愈發(fā)重要和迫切，作為實(shí)時(shí)數(shù)倉(cāng)的兩大核心組件，可以縮短數(shù)據(jù)導(dǎo)入、方便數(shù)據(jù)行級(jí)變更、支持?jǐn)?shù)據(jù)流式讀取等。

本文轉(zhuǎn)載自微信公眾號(hào)「數(shù)據(jù)社」，可以通過(guò)以下二維碼關(guān)注。轉(zhuǎn)載本文請(qǐng)聯(lián)系數(shù)據(jù)社公眾號(hào)。