一、背景

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

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

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

目前Paimon在攜程商旅的使用場景主要有以下兩個方面：

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

2.1.3 基于Lookup Join退化維度信息

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

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

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

2.2 機票自動退票提醒優(yōu)化

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

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

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

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

2.3 廣告訂單歸因準(zhǔn)實時上報

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

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

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

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

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

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

詳細過程如下：

1）ODS層構(gòu)建

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

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

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

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

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

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

4）Paimon的Catalog消費實踐

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

三、批流一體實踐

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

具體過程：

1）配置Paimon catalog

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

/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ù)平臺已默認(rèn)配置，使用方式和體驗與Spark SQL無異；

2）流批讀寫Paimon表

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

3）基于Tag的增量計算

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

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

四、總結(jié)

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

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

五、未來規(guī)劃

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