一、背景

攜程商旅專注于為企業客戶提供一站式差旅管理服務，覆蓋機票、酒店、用車、火車等多類差旅場景。用戶可通過商旅平臺完成預訂、審批、報銷等全流程操作，業務鏈條長、數據流轉環節多，數據規模與復雜度持續攀升。

伴隨商旅業務的增長和產品形態的日益豐富，業務對數據時效性的要求不斷提高，原有的T+1離線數倉架構已無法滿足準實時數據分析需求，而基于Kafka、Flink的傳統實時數倉雖能支撐部分實時計算場景，但適用性有限，且其計算中間層無法直接用于分析。

因此商旅大數據團隊積極探索準實時湖倉一體路線，提升業務數據新鮮度，推動以Paimon為代表的新型湖倉引擎在核心業務場景落地，助力數據架構升級和業務創新。

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

1）準實時數倉搭建：基于Flink CDC和Paimon搭建準實時湖倉，也是當前業界比較典型的湖倉一體解決方案；

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

二、準實時數倉搭建

2.1 準實時訂單寬表開發

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

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

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

2）而Flink+Kafka多流Join在復雜鏈路下穩定性不足，維護成本高。

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

基于Paimon的準實時寬表構建過程如下圖所示，ODS層通過Flink CDC將MySQL業務數據實時入湖，EDW層借助Paimon的Partial Update和Aggregation合并引擎構建寬表，另外也使用Paimon表當作維表存儲，代替HBase/Redis進行Lookup Join。

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

2.1.1 基于Partial Update的構建訂單產品信息寬表

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

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

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

2.1.2 基于Aggregation構建訂單中間寬表

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

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

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

2.1.3 基于Lookup Join退化維度信息

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

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

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

2.2 機票自動退票提醒優化

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

原有鏈路是T-1離線任務，提前計算第二天需提醒的訂單，下游通過獲取昨日分區數據來滿足當天的提醒需求。

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

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

2.3 廣告訂單歸因準實時上報

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

在實際落地過程中，面臨以下挑戰：  

1）上報所需字段和邏輯在業務系統中涉及7張MySQL表，實時多流Join實現難度和成本較大、穩定性挑戰較大。

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

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

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

詳細過程如下：

1）ODS層構建

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

2）基于Aggregation For Partial Update構建寬表

在訂單管理寬表構建的場景中，我們使用Partial Update打寬具有相同主鍵流表，在Partial Update的合并過程中也支持aggragation函數。在訂單上報場景的寬表實現邏輯上，上報的酒店價格需要減去所有商家側的優惠金額，涉及商家促銷表和商家優惠券表。如果需要獲取訂單的促銷優惠金額需要按照訂單號sum，因此在使用Partial Update構建寬表時使用sum聚合函數，對于促銷表和優惠券表這兩個流表只需要篩選出商家側的記錄參與寬表的構建，即可計算商家促銷優惠金額和商家優惠券金額，下圖展示了核心SQL邏輯及算子DAG流程。

3）分區數據過期機制

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

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

4）Paimon的Catalog消費實踐

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

三、批流一體實踐

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

具體過程：

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;
"

攜程大數據平臺已默認配置，使用方式和體驗與Spark SQL無異；

2）流批讀寫Paimon表

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

3）基于Tag的增量計算

Paimon 與其他數據湖技術一樣，支持 Tag 功能。Tag 是基于 Paimon 表快照創建的標簽，能夠長期保留指定快照及其對應的數據文件。Paimon 支持查詢任意兩個 Tag 之間的增量數據。結合前述結論，可以將創建時間作為分區字段，定期創建 Tag 以形成數據切片。下面是我們按天周期創建ods表的Tag切片，用于下游增量計算。

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

四、總結

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

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

五、未來規劃

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