一、背景

攜程商旅專注于為企業(yè)客戶提供一站式差旅管理服務(wù)，覆蓋機(jī)票、酒店、用車、火車等多類差旅場(chǎng)景。用戶可通過(guò)商旅平臺(tái)完成預(yù)訂、審批、報(bào)銷等全流程操作，業(yè)務(wù)鏈條長(zhǎng)、數(shù)據(jù)流轉(zhuǎn)環(huán)節(jié)多，數(shù)據(jù)規(guī)模與復(fù)雜度持續(xù)攀升。

伴隨商旅業(yè)務(wù)的增長(zhǎng)和產(chǎn)品形態(tài)的日益豐富，業(yè)務(wù)對(duì)數(shù)據(jù)時(shí)效性的要求不斷提高，原有的T+1離線數(shù)倉(cāng)架構(gòu)已無(wú)法滿足準(zhǔn)實(shí)時(shí)數(shù)據(jù)分析需求，而基于Kafka、Flink的傳統(tǒng)實(shí)時(shí)數(shù)倉(cāng)雖能支撐部分實(shí)時(shí)計(jì)算場(chǎng)景，但適用性有限，且其計(jì)算中間層無(wú)法直接用于分析。

因此商旅大數(shù)據(jù)團(tuán)隊(duì)積極探索準(zhǔn)實(shí)時(shí)湖倉(cāng)一體路線，提升業(yè)務(wù)數(shù)據(jù)新鮮度，推動(dòng)以Paimon為代表的新型湖倉(cāng)引擎在核心業(yè)務(wù)場(chǎng)景落地，助力數(shù)據(jù)架構(gòu)升級(jí)和業(yè)務(wù)創(chuàng)新。

目前Paimon在攜程商旅的使用場(chǎng)景主要有以下兩個(gè)方面：

1）準(zhǔn)實(shí)時(shí)數(shù)倉(cāng)搭建：基于Flink CDC和Paimon搭建準(zhǔn)實(shí)時(shí)湖倉(cāng)，也是當(dāng)前業(yè)界比較典型的湖倉(cāng)一體解決方案；

2）批流一體融合實(shí)踐：基于Paimon的流批一體存儲(chǔ)基礎(chǔ)上，分別用Flink和Spark進(jìn)行流式處理和批處理。

二、準(zhǔn)實(shí)時(shí)數(shù)倉(cāng)搭建

2.1 準(zhǔn)實(shí)時(shí)訂單寬表開發(fā)

訂單明細(xì)寬表是商旅訂單管理模塊的核心應(yīng)用層表，支撐用戶隨時(shí)查詢訂單明細(xì)，當(dāng)前鏈路采用小時(shí)級(jí)離線任務(wù)加工寬表。隨著商旅業(yè)務(wù)的快速發(fā)展，用戶對(duì)數(shù)據(jù)的實(shí)時(shí)性提出了更高的期望，但是訂單明細(xì)寬表字段很多，數(shù)據(jù)來(lái)源分散，ETL過(guò)程涉及近十張表、加工邏輯復(fù)雜而且鏈路較長(zhǎng)。 

在引入Paimon之前，我們也嘗試過(guò)基于Flink+Kafka搭建實(shí)時(shí)寬表，但在實(shí)際過(guò)程中暴露出以下主要痛點(diǎn)：

1）離線小時(shí)級(jí)的批任務(wù)運(yùn)行不穩(wěn)定，ETL流程一旦超時(shí)將阻塞下個(gè)小時(shí)實(shí)例運(yùn)行，數(shù)據(jù)延遲更高。

2）而Flink+Kafka多流Join在復(fù)雜鏈路下穩(wěn)定性不足，維護(hù)成本高。

如何在保證數(shù)據(jù)新鮮度的同時(shí)，兼顧開發(fā)效率和鏈路穩(wěn)定性，成為準(zhǔn)實(shí)時(shí)訂單寬表開發(fā)的核心挑戰(zhàn)。引入Paimon能夠有效解決上述問(wèn)題，其湖倉(cāng)一體的特性支持Upsert更新和動(dòng)態(tài)寫入，兼容離線與實(shí)時(shí)場(chǎng)景，Partial Update特性可代替多流Join構(gòu)建寬表，顯著提升了鏈路的穩(wěn)定性和開發(fā)效率。

基于Paimon的準(zhǔn)實(shí)時(shí)寬表構(gòu)建過(guò)程如下圖所示，ODS層通過(guò)Flink CDC將MySQL業(yè)務(wù)數(shù)據(jù)實(shí)時(shí)入湖，EDW層借助Paimon的Partial Update和Aggregation合并引擎構(gòu)建寬表，另外也使用Paimon表當(dāng)作維表存儲(chǔ)，代替HBase/Redis進(jìn)行Lookup Join。

在離線ETL任務(wù)中，寬表的加工過(guò)程通過(guò)多表Join的方式放在一個(gè)Job里完成，但Paimon的Partial Update不同于Join，使用場(chǎng)景是有條件的，要求目標(biāo)寬表的主鍵和源表主鍵相同，因此離線ETL邏輯不能照搬到實(shí)時(shí)任務(wù)上，所以將離線作業(yè)拆分為三個(gè)Flink作業(yè)：基于Partial Update構(gòu)建訂單產(chǎn)品通用信息寬表、基于Aggregation構(gòu)建訂單中間寬表、基于Lookup Join退化維度信息。

2.1.1 基于Partial Update的構(gòu)建訂單產(chǎn)品信息寬表

火車票產(chǎn)品信息表、機(jī)票產(chǎn)品信息表、產(chǎn)品通用信息表具有相同粒度的主鍵（col1,col2），具體實(shí)現(xiàn)過(guò)程中，首先創(chuàng)建一張Paimon寬表，merge-engine設(shè)置為partial-update，并通過(guò)sequence group機(jī)制控制多個(gè)流中每個(gè)流的更新順序，最終匯聚成一張訂單產(chǎn)品寬表。

下圖展示了核心SQL邏輯及算子DAG流程。

與Flink多流Join方案相比，Paimon的Partial Update機(jī)制在寬表構(gòu)建中具備明顯優(yōu)勢(shì)。首先Partial Update無(wú)需維護(hù)復(fù)雜Join產(chǎn)生的state，極大降低了作業(yè)的state存儲(chǔ)開銷，避免了因state膨脹導(dǎo)致的資源瓶頸和性能下降。其次作業(yè)的CheckPoint過(guò)程更加輕量，提升了整體鏈路的穩(wěn)定性和恢復(fù)能力，減少了因state不一致或CheckPoint失敗引發(fā)的異常。通過(guò)將多流數(shù)據(jù)的字段級(jí)變更直接落地到Paimon表，既保證了數(shù)據(jù)新鮮度，也簡(jiǎn)化了準(zhǔn)實(shí)時(shí)鏈路的開發(fā)與運(yùn)維，助力準(zhǔn)實(shí)時(shí)訂單寬表加工鏈路高效、穩(wěn)定。

2.1.2 基于Aggregation構(gòu)建訂單中間寬表

針對(duì)ODS表主鍵不一致、無(wú)法通過(guò)一次Partial Update實(shí)現(xiàn)多流數(shù)據(jù)合并的場(chǎng)景，我們采用了Paimon的Aggregation合并引擎，并結(jié)合nested_update函數(shù)進(jìn)行處理。

具體做法是：將三個(gè)主鍵分別為col1、(co1, col8)、(col1, col18) 的流表，通過(guò)Aggregation引擎聚合到以 col1 為主鍵的寬表。nested_update函數(shù)的作用類似于hive SQL中的collect_list()，能夠?qū)⒎?col1 作為主鍵的流表記錄，按  col1 聚合為Array類型，統(tǒng)一寬表的主鍵粒度。此外，對(duì)于 col8 和 col18 的計(jì)數(shù)需求，由于Paimon Aggregation引擎表暫不支持count函數(shù)，我們通過(guò)sum+case when的方式實(shí)現(xiàn)等價(jià)計(jì)算，滿足了業(yè)務(wù)對(duì)多維度數(shù)據(jù)聚合的需求。

下圖展示了核心SQL邏輯及算子DAG流程，這樣既保證了數(shù)據(jù)的完整性和一致性，也提升了寬表加工的靈活性和擴(kuò)展能力。 

2.1.3 基于Lookup Join退化維度信息

傳統(tǒng)實(shí)時(shí)數(shù)倉(cāng)中，實(shí)時(shí)場(chǎng)景Lookup Join的維表存儲(chǔ)通常選擇HBase、Redis和MySQL，它們都需要依賴第三方存儲(chǔ)，增加實(shí)時(shí)鏈路的復(fù)雜度和運(yùn)維成本。引入Paimon后，用Paimon表來(lái)存儲(chǔ)維度數(shù)據(jù)，不再依賴第三方存儲(chǔ)，而且維表數(shù)據(jù)量不大的情況下Lookup Join性能完全可以接受，大大簡(jiǎn)化了實(shí)時(shí)鏈路的架構(gòu)。

通過(guò)Aggregation加工的寬表和維表進(jìn)行Lookup Join豐富維度信息，nested_update函數(shù)聚合的字段通過(guò)unnest展開與維表Join，作用等價(jià)于常用的explode函數(shù)。

下圖展示了核心SQL邏輯及算子DAG流程。

2.2 機(jī)票自動(dòng)退票提醒優(yōu)化

機(jī)票自動(dòng)退票提醒功能要求提供當(dāng)天需提醒的機(jī)票訂單，雖然這些訂單都是歷史數(shù)據(jù)，但由于票號(hào)狀態(tài)會(huì)不斷刷新，狀態(tài)變化直接影響訂單是否需要被篩選提醒。

原有鏈路是T-1離線任務(wù)，提前計(jì)算第二天需提醒的訂單，下游通過(guò)獲取昨日分區(qū)數(shù)據(jù)來(lái)滿足當(dāng)天的提醒需求。

這種設(shè)計(jì)存在數(shù)據(jù)延遲問(wèn)題：數(shù)據(jù)延遲超過(guò)2天，雖然需提醒的訂單在近2天內(nèi)出現(xiàn)的概率極小，但實(shí)際上這段時(shí)間內(nèi)訂單票號(hào)仍可能發(fā)生變化，影響最終篩選結(jié)果。為提升數(shù)據(jù)準(zhǔn)確性和新鮮度，我們基于Flink和Paimon對(duì)原有鏈路進(jìn)行了改造。在改造過(guò)程中也發(fā)現(xiàn)，如果全鏈路僅依賴Flink實(shí)時(shí)計(jì)算，歷史數(shù)據(jù)在首次流式消費(fèi)后已被處理，后續(xù)即便滿足提醒條件但未發(fā)生數(shù)據(jù)變更，仍無(wú)法再次觸發(fā)計(jì)算，導(dǎo)致部分訂單可能被遺漏，無(wú)法及時(shí)捕獲和提醒。

在確保數(shù)據(jù)準(zhǔn)確性的基礎(chǔ)上，為提升數(shù)據(jù)新鮮度，我們?cè)O(shè)計(jì)了如圖所示的實(shí)時(shí)與離線混合鏈路：訂單票號(hào)等核心字段的加工使用Flink+Paimon準(zhǔn)實(shí)時(shí)鏈路完成，最終的訂單篩選則通過(guò)Spark批作業(yè)定時(shí)執(zhí)行，產(chǎn)出的結(jié)果表通過(guò)攜程內(nèi)部DaaS服務(wù)注冊(cè)為API，便于下游系統(tǒng)實(shí)時(shí)獲取提醒訂單，兼顧了數(shù)據(jù)的時(shí)效性與服務(wù)的穩(wěn)定性。

2.3 廣告訂單歸因準(zhǔn)實(shí)時(shí)上報(bào)

在商旅酒店廣告投放場(chǎng)景中，需將酒店列表頁(yè)涉及廣告酒店的曝光、用戶點(diǎn)擊及下單行為準(zhǔn)實(shí)時(shí)上報(bào)給廣告主。用戶下單行為的上報(bào)需與用戶近3天內(nèi)的點(diǎn)擊日志進(jìn)行歸因匹配，只有在下單時(shí)間前3天內(nèi)存在有效點(diǎn)擊行為的訂單，才會(huì)被上報(bào)給廣告主。訂單上報(bào)的場(chǎng)景對(duì)時(shí)效性有一定要求，業(yè)務(wù)方期望能夠做到端到端分鐘級(jí)時(shí)效。

在實(shí)際落地過(guò)程中，面臨以下挑戰(zhàn)：  

1）上報(bào)所需字段和邏輯在業(yè)務(wù)系統(tǒng)中涉及7張MySQL表，實(shí)時(shí)多流Join實(shí)現(xiàn)難度和成本較大、穩(wěn)定性挑戰(zhàn)較大。

2）點(diǎn)擊日志每日增量多，數(shù)據(jù)表膨脹速度較快，需有效控制表存儲(chǔ)，保障查詢和Join性能。

如何高效整合多表數(shù)據(jù)、管理膨脹的點(diǎn)擊日志表，并滿足分鐘級(jí)別的上報(bào)時(shí)效，是該場(chǎng)景下的核心業(yè)務(wù)痛點(diǎn)。

最終設(shè)計(jì)開發(fā)的ETL鏈路如下圖所示，基于Aggregation For Partial Update解決多流join的挑戰(zhàn)，通過(guò)Append Scalable表和分區(qū)數(shù)據(jù)過(guò)期機(jī)制來(lái)提高Lookup Join的效率和穩(wěn)定性，采用Filesystem Catalog實(shí)時(shí)消費(fèi)Paimon表并同步調(diào)用SOA服務(wù)進(jìn)行數(shù)據(jù)上報(bào)。結(jié)合Flink作業(yè)3～5分鐘的Checkpoint周期，整個(gè)鏈路端到端延遲穩(wěn)定控制在8分鐘以內(nèi)。

詳細(xì)過(guò)程如下：

1）ODS層構(gòu)建

依然是借助Flink CDC全增量一體同步的功能，將MySQL數(shù)據(jù)實(shí)時(shí)入湖，需要注意的是ODS表的bucket數(shù)設(shè)置，需要估算表的大小以及考慮近幾年的數(shù)據(jù)增量，按照官方建議的每個(gè)bucket 控制在1G左右設(shè)置bucket數(shù)量。

2）基于Aggregation For Partial Update構(gòu)建寬表

在訂單管理寬表構(gòu)建的場(chǎng)景中，我們使用Partial Update打?qū)捑哂邢嗤麈I流表，在Partial Update的合并過(guò)程中也支持aggragation函數(shù)。在訂單上報(bào)場(chǎng)景的寬表實(shí)現(xiàn)邏輯上，上報(bào)的酒店價(jià)格需要減去所有商家側(cè)的優(yōu)惠金額，涉及商家促銷表和商家優(yōu)惠券表。如果需要獲取訂單的促銷優(yōu)惠金額需要按照訂單號(hào)sum，因此在使用Partial Update構(gòu)建寬表時(shí)使用sum聚合函數(shù)，對(duì)于促銷表和優(yōu)惠券表這兩個(gè)流表只需要篩選出商家側(cè)的記錄參與寬表的構(gòu)建，即可計(jì)算商家促銷優(yōu)惠金額和商家優(yōu)惠券金額，下圖展示了核心SQL邏輯及算子DAG流程。

3）分區(qū)數(shù)據(jù)過(guò)期機(jī)制

訂單歸因需要關(guān)聯(lián)訂單下單時(shí)間前3天內(nèi)的點(diǎn)擊記錄，因此點(diǎn)擊記錄維表的生命周期設(shè)置為3天。Paimon提供了兩種數(shù)據(jù)失效機(jī)制：一種是基于主鍵表的record level expiration，另一種是基于分區(qū)的partition level expiration。

我們分別對(duì)這兩種方式進(jìn)行了實(shí)踐，表配置如下圖所示，實(shí)際效果來(lái)看，記錄級(jí)失效（record level expiration）如官方文檔所述，無(wú)法保證及時(shí)清除過(guò)期數(shù)據(jù)，離預(yù)期效果相差甚遠(yuǎn)。相比之下，采用非主鍵、動(dòng)態(tài)分桶的分區(qū)表，并設(shè)置分區(qū)保留4天（partition expiration），能夠確保分區(qū)最早日期超過(guò)4天時(shí)自動(dòng)失效，Lookup Join過(guò)程始終可關(guān)聯(lián)到下單時(shí)近3天的點(diǎn)擊日志。這種實(shí)現(xiàn)方式的DAG流程如下圖所示，也是官方推薦的實(shí)現(xiàn)方式，支持自動(dòng)compaction合并小文件，能有效控制數(shù)據(jù)量規(guī)模并提升查詢效率。

4）Paimon的Catalog消費(fèi)實(shí)踐

點(diǎn)擊記錄和訂單歸因結(jié)果寫入Paimon表后，需要同時(shí)調(diào)用廣告投放方的SOA服務(wù)進(jìn)行上報(bào)，因此服務(wù)調(diào)用需集成進(jìn)整個(gè)準(zhǔn)實(shí)時(shí)鏈路。結(jié)合官方文檔提供的多種Catalog類型，考慮到內(nèi)部權(quán)限和認(rèn)證問(wèn)題，最終選擇了訪問(wèn)便捷的Filesystem Catalog，將訂單歸因結(jié)果表注冊(cè)為DataStream流，同時(shí)調(diào)用下游SOA服務(wù)完成上報(bào)，既保證了數(shù)據(jù)處理的時(shí)效性，也簡(jiǎn)化了鏈路的權(quán)限管理和運(yùn)維復(fù)雜度。

三、批流一體實(shí)踐

現(xiàn)階段流批一體方向由于Flink的批處理能力無(wú)法代替Spark，尤其是SQL語(yǔ)義的差異較大，所以暫時(shí)不能做到計(jì)算引擎和代碼層面的流批一體。當(dāng)前比較成熟和落地的場(chǎng)景是流批一體存儲(chǔ)，即Flink CDC流式寫入Paimon后，基于相同的Paimon ODS表Spark負(fù)責(zé)批處理、Flink負(fù)責(zé)流處理，整體仍然是Lambda架構(gòu)。

具體過(guò)程：

1）配置Paimon catalog

Spark3可以通過(guò)catalog讀寫Paimon表，配置過(guò)程如下：

/opt/app/spark-3.2.0/bin/spark-sql \
  --conf 'spark.sql.catalog.paimon_catalog=org.apache.paimon.spark.SparkGenericCatalog' \
  --conf 'spark.sql.extensinotallow=org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions' \
  --conf 'spark.sql.storeAssignmentPolicy=ansi' \
  -e "
select * from paimon.dp_lakehouse.ods_xxx limit 10;
"

攜程大數(shù)據(jù)平臺(tái)已默認(rèn)配置，使用方式和體驗(yàn)與Spark SQL無(wú)異；

2）流批讀寫Paimon表

得益于 Paimon 對(duì)主鍵表的 Upsert 及動(dòng)態(tài)分區(qū)寫入能力，流批 ETL 鏈路具備了實(shí)現(xiàn)增量計(jì)算的基礎(chǔ)。實(shí)際應(yīng)用中，我們分別嘗試以創(chuàng)建時(shí)間和更新時(shí)間作為分區(qū)字段：以創(chuàng)建時(shí)間分區(qū)，可實(shí)現(xiàn)動(dòng)態(tài)分區(qū)的更新，支持歷史數(shù)據(jù)的回溯更新，但難以高效掃描變更數(shù)據(jù)；以更新時(shí)間分區(qū)，能夠便捷獲取變更數(shù)據(jù)，但不支持歷史分區(qū)的數(shù)據(jù)回溯更新。因此，如何在高效獲取增量數(shù)據(jù)的同時(shí)，兼顧歷史數(shù)據(jù)的更新能力，是實(shí)現(xiàn)增量計(jì)算的關(guān)鍵挑戰(zhàn)。

3）基于Tag的增量計(jì)算

Paimon 與其他數(shù)據(jù)湖技術(shù)一樣，支持 Tag 功能。Tag 是基于 Paimon 表快照創(chuàng)建的標(biāo)簽，能夠長(zhǎng)期保留指定快照及其對(duì)應(yīng)的數(shù)據(jù)文件。Paimon 支持查詢?nèi)我鈨蓚€(gè) Tag 之間的增量數(shù)據(jù)。結(jié)合前述結(jié)論，可以將創(chuàng)建時(shí)間作為分區(qū)字段，定期創(chuàng)建 Tag 以形成數(shù)據(jù)切片。下面是我們按天周期創(chuàng)建ods表的Tag切片，用于下游增量計(jì)算。

通過(guò) Tag 之間的增量查詢，不僅能夠高效獲取數(shù)據(jù)變更，還能將增量計(jì)算數(shù)據(jù)寫入目標(biāo)分區(qū)表，實(shí)現(xiàn)對(duì)歷史數(shù)據(jù)的回溯更新。該方案在批處理場(chǎng)景下的增量計(jì)算具有重要意義，不僅能夠節(jié)省ETL的計(jì)算資源，還大幅縮短了作業(yè)執(zhí)行時(shí)間。在我們內(nèi)部實(shí)踐中，基于Tag的增量計(jì)算替代全量ETL后，作業(yè)的處理速度提升了4～5倍，尤其在增量數(shù)據(jù)較少的ETL場(chǎng)景下，帶來(lái)了顯著收益。

四、總結(jié)

如本文所述，基于 Flink CDC 與 Paimon 的準(zhǔn)實(shí)時(shí)數(shù)倉(cāng)架構(gòu)，有效支撐了攜程商旅多個(gè)場(chǎng)景的準(zhǔn)實(shí)時(shí)數(shù)據(jù)應(yīng)用需求。通過(guò)主鍵表 Upsert 替代 Row_number() 去重，利用 Aggregation 聚合函數(shù)代替 SQL 中的 Group By 操作，有效提升了鏈路效率。對(duì)于寬表與流表主鍵粒度一致的場(chǎng)景，優(yōu)先采用 Partial Update 方式構(gòu)建寬表，實(shí)現(xiàn)高效的數(shù)據(jù)合并與更新，若主鍵粒度不一致，則采用 Aggregation的 Nested_update 和 Unnest 組合，靈活滿足多樣化的數(shù)據(jù)整合需求。在性能開銷方面，Partial Update 優(yōu)于Lookup Join，Lookup join又優(yōu)于 Regular Join，整體方案兼顧了實(shí)時(shí)性、查詢效率與運(yùn)維簡(jiǎn)易性，顯著提升了業(yè)務(wù)支撐時(shí)效性。

此外 Paimon 的 Tag 功能在批處理場(chǎng)景下的增量計(jì)算中具有重要應(yīng)用價(jià)值。通過(guò)基于快照創(chuàng)建 Tag，可以定期對(duì)數(shù)據(jù)進(jìn)行切片，長(zhǎng)期保留關(guān)鍵時(shí)間點(diǎn)的歷史數(shù)據(jù)。利用 Tag 之間的增量查詢能力，能夠高效獲取數(shù)據(jù)變更，實(shí)現(xiàn)批量場(chǎng)景下的高效數(shù)據(jù)同步與回溯更新。這不僅顯著提升了計(jì)算效率，還增強(qiáng)了數(shù)據(jù)的可維護(hù)性和靈活性。

五、未來(lái)規(guī)劃

當(dāng)前業(yè)務(wù)實(shí)踐仍采用 Lambda 架構(gòu)，計(jì)算與存儲(chǔ)分離。出于業(yè)務(wù)穩(wěn)定性的考量，暫未在實(shí)時(shí)場(chǎng)景中實(shí)踐 Branch 和 Tag 等特性。后續(xù)將重點(diǎn)探索 Paimon 與 Flink 的流批一體能力，進(jìn)一步推動(dòng)計(jì)算與存儲(chǔ)的深度融合。