摘要：在 Flink Forward Asia 大會實時數倉專場中，菜鳥資料&規劃部進階資料技術專家賈元喬從資料模型、資料計算、資料服務等幾個方面介紹了菜鳥供應鍊資料團隊在實時資料技術架構上的演進，以及在供應鍊場景中典型的實時應用場景和 Flink 的實作方案。
首先從三個方面簡要介紹一下菜鳥早在 2016 年采用的實時資料技術架構：資料模型、實時計算和資料服務。
- 資料模型。菜鳥最初使用的是需求驅動的、縱向煙囪式的開發模式，計算成本高且完全沒有複用的可能性，同時也會導緻資料一緻性的問題；整個資料模型沒有分層，業務線内部模型層次混亂，使得資料使用成本特别高。
- 實時計算。該部分使用的是阿裡的 JStorm 和 Spark Streaming，大多數情況下，二者可以滿足實時計算的需求，但是對于有些複雜的功能，如物流和供應鍊場景，實作起來不夠簡單，開發成本較高；同時很難兼顧功能、性能、穩定性以及快速的故障恢複能力。
- 資料服務。資料主要存儲在 Hbase、MySQL 和 ADB 等不同類型的資料庫中，然而對于很多營運人員來說，查詢資料庫的頻率并不高，但使用資料庫的成本較高，尤其針對一些 NoSQL 的資料庫；也存在資料使用不可控，如熱點阻斷、權限控制以及全鍊路監控等問題。
針對以上問題，菜鳥在 2017 年對資料技術架構進行了一次比較大的更新改造，以下将詳細介紹。
資料模型更新
資料模型的更新主要是模型分層，充分複用公共中間層模型。之前的模式是從資料源 TT（如 Kafka）中抽取資料并進行加工，産生一層式的表結構。新版本的資料模型進行了分層：
- 第一層是資料采集，支援多種資料庫中的資料采集，同時将采集到的資料放入消息中間件中；
- 第二層是事實明細層，基于TT的實時消息産生事實明細表，然後再寫入TT的消息中間件中，通過釋出訂閱的方式彙總到第三、四層，分别是輕度彙總層和高度彙總層。
- 第三層輕度彙總層适合資料次元、名額資訊比較多的情況，如大促統計分析的場景，該層的資料一般存入阿裡自研的 ADB 資料庫中，使用者可以根據自己的需求篩選出目标名額進行聚合；
- 而第四層高度彙總層則沉澱了一些公共粒度的名額，并将其寫入 Hbase 中，支援大屏的實時資料顯示場景，如媒體大屏、物流大屏等。
原本采用的開發模式各個業務線獨立開發，不同業務線之間不考慮共性的問題，但物流場景中，很多功能需求其實是類似的，這樣往往會造成資源的浪費，針對該問題進行的改造首先是抽象出橫向的公共資料中間層（左側藍色），然後各個業務線在此基礎上分流自己的業務資料中間層（右側黃色）。
前面介紹的業務線分流由預置的公共分流任務來實作，即将原來下遊做的分流作業，全部轉移到上遊的一個公共分流作業來完成，充分複用公共預置分流模型，大大節省計算資源。
下面介紹一個資料模型更新的具體案例—菜鳥供應鍊實時資料模型。
- 下圖左側是前面介紹的公共資料中間層，包括整個菜鳥橫向的物流訂單、大盤物流詳情和公共粒度的一些資料，在此基礎上菜鳥實作了預置公共分流，從物流訂單、物流詳情中拆分出個性化業務線的公共資料中間層，包括國内供應鍊、進口供應鍊以及出口供應鍊等。
- 基于已經分流出來的公共邏輯，再加上業務線個性化TT的消息，産出各業務線的業務資料中間層。
- 以進口供應鍊為例，其可能從公共業務線中分流出物流訂單和物流詳情，但是海關資訊、幹線資訊等都在自己的業務線進口供應鍊的TT中，基于這些資訊會産生該業務線的業務資料中間層。
借助前面所述的設計理念，再加上實時的模型設計規範和實時的開發規範，大大提升了資料模型的易用性。
計算引擎更新
菜鳥最初的計算引擎采用的是阿裡内部研發的 JStorm 和 Spark Streaming，可以滿足大多數場景的需求，但針對一些複雜的場景，如供應鍊、物流等，會存在一些問題。是以，菜鳥在 2017 年全面更新為基于 Flink 的實時計算引擎。當時選擇 Flink 的主要原因是：
- Flink 提供的很多功能非常适用于解決供應鍊場景下的需求，菜鳥内部提煉了一套 Flink 的 SQL 文法，簡單易用且标準化，大大提升了開發效率。
- 此外，Flink 内置的基于 state 的 Retraction 的機制可以很好地支援供應鍊場景下的取消訂單、換配需求的實作；
- 後來推出的 CEP 功能使得物流、供應鍊中實時逾時統計需求的實作變得更加簡單；
- AutoScaling 等自動優化的方案可以使得菜鳥省去了一些資源配置等方面的複雜性和成本；
- 半智能功能如批流混合等也較好地滿足菜鳥業務的實際需求。
下面介紹三個與計算引擎更新相關的案例。
案例 1：基于 state 的 Retraction
下圖左側是一個物流訂單表，包含了四列資料，即物流訂單号、建立時間、是否取消和計劃配送公司。假設有一個需求是統計某個配送公司計劃履行的有效單量是多少，該需求看起來簡單，實際實作過程中有有一些問題需要注意。
- 一個問題是針對表中 LP3 訂單，在開始的時候是有效的（18 分的時候“是否取消”應該是 N，表寫錯），然而最後該訂單卻被取消了（最後一行“是否取消”應該是Y，表寫錯）,這種情況該訂單被視為無效訂單，統計的時候不應該考慮在内。
- 另外，配送公司的轉變也需要注意，LP1 訂單在 1 分鐘的時候計劃配送公司還是 tmsA，而之後計劃配送公司變成了 tmsB 和 tmsC，按照離線的計算方式（如 Storm 或增量）會得出右上角的結果，tmsA、tmsB 和 tmsC 與 LP1 訂單相關的記錄都會被統計，事實上 tmsA 和 tmsB 都未配送該訂單，是以該結果實際上是錯誤的，正确的結果應該如圖右下角表格所示。
針對該場景，Flink 内置提供了基于 state 的 Retraction 機制，可以幫助輕松實作流式消息的回撤統計。
下圖展示了 Retraction 機制的僞代碼實作。第一步是利用 Flink SQL 内置行數 last_value，擷取聚合 key 的最後一條非空的數值，針對上述表中的 LP1 訂單，使用 last_value 得到的結果是 tmsC，是符合預期的結果。需要強調的一點是，左側使用 last_value 統計的字段 gmt_create、plan_tms、is_cancel，一旦其中的任何一個字段發生變化，都會發生出發 Flink 的 Retraction 機制。
案例 2：逾時統計
物流是菜鳥中比較常見的業務場景，物流業務中經常會有實時逾時統計的需求，比如統計倉出庫超過六個小時未被攬收的單量。
- 用到的資料表如下圖左側所示，其中包含日志時間、物流訂單号、出庫時間和攬收時間。該需求如果在離線的小時表或天表中比較好實作，但是在實時的場景下，其實作面臨一定的挑戰。
- 因為如果倉出庫後未被攬收，意味着沒有新的消息流入，如果沒有消息就沒有辦法進行逾時消息的計算。
- 為了解決該問題，菜鳥從 2017 年初就開始了一系列的探索，發現一些消息中間件（如 Kafka）和 Flink CEP 等本身會提供逾時消息下發的功能，引入消息中間件的維護成本比較高，而 Flink CEP 的應用會出現回傳不準确的問題。
針對上述需求，菜鳥選擇了 Flink Timer Service 來進行實作。具體來講，菜鳥對 Flink 底層的 ProcessFunction 中的 ProcessElement 函數進行了改寫,該函數中，由 Flink 的 state 存儲原始消息，相同的主鍵隻存一次，一旦 endNode 已實操，則 state 消息置為無效，已逾時的消息直接下發。此外，重寫編寫一個 OnTimer 函數，主要負責在每個逾時的時刻讀取 state 消息，然後下發 state 中仍然有效的消息，基于下遊和正常遊的關聯操作便可以統計出逾時消息的單量。
使用 Flink Timer Service 進行逾時統計的僞代碼實作如下圖所示。
- 首先需要建立執行環境，構造 Process Function（通路 keyed state 和 times）；
- 其次是 processElement 函數的編寫，主要用于告訴 state 存儲什麼樣的資料，并為每個逾時消息注冊一個 timerService，代碼中 timingHour 存儲逾時時間，比如前面的提到六小時，
- 然後啟動 timerService；
- 最後是 onTimer 函數的編寫，作用是在逾時的時刻讀取 state 的資料，并将逾時消息下發。
案例 3：從手動優化到智能優化
實時數倉中會經常遇到資料熱點和資料清洗的問題。下圖左側展示了資料熱點的流程，藍色部分 Map 階段經過 Shuffle 後，轉到紅色部分 Agg，此時便會出現資料熱點。針對該問題，菜鳥最初的解決方案的僞代碼實作如下圖右側所示。假設對 lg_order-code 進行清洗，首先會對其進行 hash 散列操作，然後針對散列的結果進行二次聚合，這樣便可以在一定程度上減輕傾斜度，因為可能會多一個 Agg 的操作。
菜鳥内部目前使用的 Flink 最新版本提供了解決資料熱點問題的智能化特性：
- MiniBatch。原來每進來一條資料，就需要去 state 中查詢并寫入，該功能可以将資料進行聚合後再寫入 state 或從 state 中讀取，進而減輕對 state 的查詢壓力。
- LocalGlobal。類似于 Hive 中 Map 階段的聚合，通過該參數可以實作資料讀取階段的聚合，輕松應對 count 熱點。
- PartialFinal。面對更複雜的場景，比如 count_distinct 的熱點，使用該參數可以輕松應對，實作兩次聚合，類似于 Hive 中的兩次 Reduce 操作。
智能化功能支援的另一個場景是資源配置。在進行實時 ETL 過程中，首先要定義 DDL，然後編寫 SQL，之後需要進行資源配置。針對資源配置問題，菜鳥之前的方案是對每一個節點進行配置，包括并發量、是否會涉及消息亂序操作、CPU、記憶體等，一方面配置過程非常複雜，另一方面無法提前預知某些節點的資源消耗量。Flink目前提供了較好的優化方案來解決該問題：
- 大促場景：該場景下，菜鳥會提前預估該場景下的 QPS，會将其配置到作業中并重新開機。重新開機後 Flink 會自動進行壓測，測試該 QPS 每個節點所需要的資源。
- 日常場景：日常場景的 QPS 峰值可能遠遠小于大促場景，此時逐一配置 QPS 依然會很複雜。為此 Flink 提供了 AutoScaling 智能調優的功能，除了可以支援大促場景下提前設定 QPS 并壓測擷取所需資源，還可以根據上遊下發的 QPS 的資料自動預估需要的資源。大大簡化了資源配置的複雜度，使得開發人員可以更好地關注業務邏輯本身的開發。
資料服務更新
菜鳥在做數倉的過程中也會提供開發一系列的資料産品來提供資料服務，原來是采用 Java Web 提供多種連接配接 DB 的方式。但是實際應用過程中，經常用到的資料庫無非是 Hbase、MySQL 和 OpenSearch 等，是以後來菜鳥聯合資料服務團隊建立了一個統一的資料服務中間件“天工資料服務”。它可以提供統一的資料庫接入、統一的權限管理、統一的資料保障以及統一的全鍊路監控等中心化的功能，将 SQL 作為一等公民，作為資料服務的 DSL，提供标準化的服務接入方式（HSF）。
作為菜鳥資料服務的踐行者，天工還提供了很多貼近業務的功能。接下來通過具體的案例場景來介紹。
案例 1：NoSQL to TgSQL
Hbase 等 NoSQL 類型的資料庫，對于營運人員來講編寫代碼是比較困難的，這種情況下其急需一套标準的文法。為了解決該問題，天工提供了 TgSQL，用于标準化 NoSQL 的轉換。下圖展示了轉換的過程，Employee 轉換成一個二維表，這裡隻是邏輯轉換而非實體轉換。天工中間件會解析 SQL，并在背景自動轉換成查詢的語言對資料進行查詢。
案例 2：跨源資料查詢
菜鳥在開發資料産品的過程中，會經常遇到實時和離線分不開的情況。比如菜鳥每年都會統計 KPI 的實時完成率，計算方式是已經完成的單量與計劃單量之間的比值，該計算依賴的資料源有兩部分：
- 一部分是已經計劃好的離線 KPI 表；
- 另一部分是已經計算好的寫入 Hbase 的實時表。
原本的實作方案是通過 Java 取兩次接口，然後在前端進行加減乘除的計算操作。針對該問題，菜鳥提供了标準的 SQL，用針對跨資料源的查詢，如 MySQL 離線表和 Hbase 實時表，使用者隻需要按照标準 SQL 的方式來寫，通過更新的資料服務進行解析，再從對應的資料庫中進行資料的查詢操作。
案例3：服務保障更新
菜鳥最初對于服務的保障比較缺失，一個任務釋出後并不确定是否有問題，有些問題直到使用者回報的時候才能發現。另外，當并發量比較大的時候，也沒有辦法及時地做限流和主備切換等應對措施。
為此，天工的中間件提供了資料保障功能，除了主備切換，還包括主備雙活、動态負載、熱點服務阻斷以及白名單限流等功能。
- 對于主備切換，前面提到的左右兩側分别是實體表和邏輯表的場景中，一個邏輯表可以映射成主備鍊路，當主鍊路出現問題時，可以一鍵切換到備鍊上；
- 此外，大促期間一些非常重要的業務，如大屏業務、内部統計分析等，會通過主備鍊路同時進行操作，此時完全讀寫其中一個庫不合适，所期望的兩條鍊路均有流量，而天工則實作了主備雙活的功能支援，即将大流量切到主鍊，小流量切到備鍊；
- 當主鍊上受到其中一個任務影響時，該任務會被移到備鍊上；對于比較複雜、執行較慢的查詢，會對整個任務的性能造成影響，此時會對這種類型的熱點服務進行阻斷。
其他技術工具的探索和創新
除了資料模型、計算引擎和資料服務，菜鳥還在其他方面進行了探索和創新，包括實時壓測、過程監控、資源管理和資料品質等。
實時壓測在大促期間比較常用，通過實時壓測來模拟大促期間的流量，測試特定的 QPS 下任務是否可以成功執行。原本的做法是重新開機備鍊上的作業，然後将備鍊作業的 source 改為壓測的 source，sink 改為壓測 source 的動作，這種方案在任務特别多的時候實作起來非常複雜。為此，阿裡雲團隊開發了實時壓測的工具，可以做到一次啟動所有的需要的壓測的作業，并自動生成壓測的 source 和 sink，執行自動壓測，生成壓測報告。
采用 Flink 後，還實作了作業過程監控的功能，包括延遲監控和告警監控，比如超過特定的時間無響應會進行告警，TPS、資源預警等。
菜鳥實時數倉未來發展與思考
菜鳥目前在實時數倉方面更多的是基于 Flink 進行一系列功能的開發，未來的發展方向計劃向批流混合以及 AI 方向演進。
（1）Flink 提供了 batch 的功能後，菜鳥很多中小型的表分析不再導入到 Hbase 中， 而是在定義 source 的時候直接将 MaxCompute 的離線維表讀到記憶體中，直接去做關聯，如此一來很多操作不需要再進行資料同步的工作。
（2）針對一些物流的場景，如果鍊路比較長，尤其是雙十一支付的訂單，在十一月十七号可能還存在未簽收的情況，這時候如果發現作業中有一個錯誤，如果重新開機的話，作業的 state 将會丢失，再加之整個上遊的 source 在 TT 中隻允許儲存三天，使得該問題的解決變得更加困難。
- 菜鳥之後發現 Flink 提供的 batch 功能可以很好地解決該問題，具體來講是定義 TT 的 source，作為三天的實時場景的應用，TT 資料寫到離線資料庫進行曆史資料備份，如果存在重新開機的情況，會讀取并整合離線的資料，即使 Flink 的 state 丢失，因為離線資料的加入，也會生成新的 state，進而不必擔心雙十一的訂單如果在十七号簽收之前重新開機導緻無法擷取十一号的訂單資訊。
- 當然，在上述問題的解決上，菜鳥也踩了很多的小坑。其中的一個是整合實時資料和離線資料的時候，資料亂序的問題。菜鳥實作了一系列的 UDF 來應對該問題，比如實時資料和離線資料的讀取優先級設定。
（3）針對日志型的業務場景，比如曝光、網站流量等，其一條日志下來後，基本不會再發生變化。菜鳥目前在考慮将所有解析的工作交給 Flink 來處理，然後再寫入到 batch 中，進而無需在 MaxCompute 的 ODPS 中進行批處理的操作。
（4）在智能化方面，前面提到的資料傾斜隐患的規避、資源的優化等，都用到了 Flink 提供的智能化功能。
- 菜鳥也期望在實時 ETL 過程中的一些場景中，比如去重，也使用 Flink 相應的智能化解決方案來進行優化。
- 此外，在資料服務保障上，如主備切換等，目前仍然依賴人工對資料庫進行監控，菜鳥也期望 Flink 之後能提供全鍊路實時保障的政策。
- 最後是業務場景的智能化，阿裡 Alink 對于業務智能化的支援也是之後探索的方向。