大家晚上好，感謝大家能來參加本次的在線分享，首先自我介紹一下，我是14年校招加入百度，之前在百度質量部參與EP相關的一些項目研發，15年加入百度外賣，主要負責數據倉庫和ETL以及部分數據產品的開發。
今天主要給大家介紹一下百度外賣這邊在ETL方向的一些實踐。整體分四塊，第一塊先介紹下ETL的需求來源，以及外賣這個業務在ETL中的一些場景特徵，這些場景特徵和後面講到的ETL系統設計聯繫比較緊密。第二部分講基於這些場景和業務，ETL都需要滿足哪些特性，都針對性的做出了什麼樣的設計。第三部分介紹了整個數據的交付體系，在數據交付給業務使用時如何保證其可用性，如何做到生產進度透明化。第四部分是介紹如何通過數據的血緣關係，來做到數據可解釋，同時基於數據血緣關係還可以做那些事，能帶來什麼樣的特性。
聊ETL自然離不開倉庫，ETL服務將數據入到倉庫，最終也都是為了在倉庫層面提供結構或半結構化的數據存儲、提供面向主題的明細和聚合數據，更好的支持上層應用及業務。
外賣這邊數據需求（也就是數據出口）主要分三大類：一，分析類需求。相信大多數業務後端RD都做過給老闆或PM跑數的需求，其實這就是最原始的ETL需求，外賣這邊每日的業務整體日報，分城市、分代理商的日報，以及各種通過“自如”自定義或訂閱的報表，都是基於ETL系統在釋放數據可用信號後開始跑相應的報表服務來實現依賴發送的。二，線上業務產品的需求。比如在面向商戶的產品中，展示商戶的流量、復購、潛在顧客等相關dashboard，都是基於倉庫的數據統計分析出來的。再比如下發給bd的商機系統，為銷售提供銷售線索的指標；面向bd的kpi考核，kpi每個指標的口徑定義，具體實現就是對應在ETL的處理邏輯上。除產品研發相關的業務外，同時支持公司內其他業務部門，如財務、預算等等。三，就是最能挖掘數據潛在價值的策略方向的應用，支撐rank、搜索推薦、畫像等。良好的結構化數據和倉庫層次劃分，能極大減小數據挖掘在前期清洗和預處理的成本，同時提供便捷的數據使用方式，讓業務部門能更加高效的專注於自己的業務領域。
針對以上業務場景，在ETL中又有著怎樣的數據場景，有哪些特別的地方呢，我們從這三個視角來看。
首先，從數據流上來講，數據主要有兩種流向，一種是從業務庫中整表抽取或部分字段抽取。整表抽取，主要是業務端的一些維度表，量級不大，變更頻率低。分字段抽取，這個是核心場景，一般的業務主體數據在OLTP場景中都會細分業務進行分庫分表，那麼對應在倉庫中的一個主題數據，比如訂單，其基礎指標在業務庫中就可能存的比較分散，比如訂單ID、價格、狀態等在一個表，訂單退款詳情、評論、投訴等信息在另一個庫另幾張表，那麼就需要完成一個跨庫跨表按訂單id抽取後數據拼裝的過程。另一種數據流向就是倉庫內部表之間的流轉，比如彙總表的指標需要用事實表和維度表進行關聯後聚合得出，再比如商戶事實表的商戶日流水指標，需要訂單表和商戶維度表關聯後聚合產出。
從邏輯視角來看，有兩種邏輯映射場景，比如價格字段，在倉庫的指標中細分為優惠前單價、優惠後單價，優惠前單價直接取業務庫訂單表中的“訂單價格”即可，這是個1對1的映射關係；優惠後單價，就需要“訂單價格”減去“商戶補貼”減去“平臺補貼”減去“支付補貼”，這是個多對一的關係。這裡就涉及到一個細節問題，加入上述三個補貼字段在業務庫的訂單表中都是以單個字段形式存在於schema中，多對一的計算直接一個select sql就能計算出來，但如果是補貼字段放在一個壓縮的json字符串中呢，就需要先解析後逐一取出再計算，就涉及到需要編碼來解決了。
從業務視角來看，數據可分為事實數據、維度數據、遲到的事實（也可理解為會變化的維度），比如訂單ID、商戶名稱，這是事實的固有屬性，是不變的事實；維度數據，比如城市、配送類型等；遲到的事實，比如訂單狀態，在訂單表中，它即是訂單的一個事實指標也是訂單的一個維度，但它是隨時間線性漸變的，也就是說在數據抽取入倉庫後，業務庫還會發生變化，會出現數據一致性的問題，這時候就涉及到數據重入和更新的問題了。在外賣這個業務中，是電商+實時物流的形態，且業務週期較長，涉及到的角色主體和業務形態眾多，如訂單、商戶、物流、用戶、騎士、bd、營銷、結算等等，可謂是訂單類的場景中最複雜的一個。
面臨以上覆雜的場景和邏輯，如果每個ETL細節邏輯都靠編碼來實現，且不說開發成本和數據解釋成本，單是任務管理和代碼繼承及足以讓開發者頭疼不已。但是每一家公司的ETL基本上都是從最初的腳本+sql的形態開始，都會經歷這個階段。
在外賣ETL方向的發展可以分為這四個階段，也是個人認為業內ETL整體發展的一個走向。從在關係型數據庫上跑sql，寫腳本解析數據，定時任務出報表開始，隨著業務體量的發展，在計算性能和任務管理方式上很快就會遇到瓶頸，單機跑腳本會逐漸由分佈式任務來代替。進入到第二階段，map-reduce、spark之類的分佈式任務和基於hadoop生態的一系列大數據解決方案實現了計算和存儲的可擴展，但面臨成千上萬的任務，早期的ct任務靠時間約定來進行上下游約束，亦或在腳本中串行調用下游來實現，這種刀耕火種的方式在業務快速膨脹的過程中，面臨的將是高昂的任務維護成本。在第三階段，引入以調度為中心的，提供監控、任務重試機制、性能分析等一整套解決方案，來實現任務的串行依賴和並行計算的靈活組合。到這一步，就已經能解決絕大多數ETL場景的問題了。在第四階段，做的就是ETL精細化，實現數據可用性監控，通過透明化數據生產進度，進而構建完整的數據交付體系，面臨龐大的元數據體系如何降低數據解釋成本，如何從表粒度的數據生產依賴，精細化到字段粒度。在這個階段，核心就是讓業務更加便捷更加自助式的使用數據，同時提升數據交付質量。
接下來看下百度外賣這邊整體數據生產的一個數據流架構。和業內所有的數據生產場景一樣，主體分為數據源、生產、倉庫、數據應用四層，數據源以業務庫的mysql為主，其次是hdfs或ftp的文件、kafka/nmq（百度的消息隊列）、api。生產層面以ETL系統為主，主要處理業務數據，日誌的基礎數據生產走spark streaming。
ETL系統的執行單元是個腳本命令，由調度系統遠程調起，來實現分佈式，調度由任務(job)->轉換(transformer)->算子(operator)的粒度進行封裝，通過這種結構關係實現ETL作業的先後依賴和並行處理。在數據生產的環節中，除通用的系統監控服務外，還開發了一套數據監控服務來保證數據產出後的可用性。
數據的例行生產由底層事實表和維度表開始，基於事實和維度生產一些中間表(也稱為輕度聚合表)，然後是聚合表，這也是常見的一種倉庫設計。就數據的使用層面來講，事實明細+維度的查詢是能滿足所有查詢場景的，靈活度也最高，但隨著業務體量的發展，這種基於明細的查詢會帶來極大的冗餘查詢。試想不同的業務查詢同一個指標，類似的sql每個人都需要跑一遍，冗餘查詢引發的將是大量集群的額外開銷。所以設計基於明細數據的聚合表，是一種事前計算的思路，將不同指標在不同維度下的數據提前計算好，方便業務查詢的同時也統一了數據口徑。在上期大數據分享中提到的kylin，其性能高效的原理和建中間表是同理，基於各種維度預建好中間表，任意維度組合的查詢都能直接落在一張表的數據文件上，極大減少了hdfs文件掃描的開銷。
那麼中間表該怎麼建，這麼多的維度組合和指標，不可能面面俱到，只能用有限的集群資源去解決熱點的問題。哪些查詢是熱點的，不可能通過逐一收集用戶需求來實現，在數據使用服務上我們開發了一套adhoc（即席查詢）查詢服務，所有的數據出口都收斂在這裡，有面向BI分析人員的sql查詢UI，也有面向後端服務的同步和異步接口，通過adhoc每天記錄的用戶查詢日誌，我們能分析出哪些表和字段是查詢熱點，都分部在什麼維度。同時通過收斂查詢的出口，在adhoc這個數據出口上建立了完整的字段粒度的數據權限控制。在ETL生產的過程中，配合信號燈服務，實現了數據生產可解釋；通過在數據集市中的元數據管理系統，支持面向用戶的元數據和口徑查詢。基於對ETL配置的解析，生產了數據的血緣關係，用於描述表的上下游及字段之間的生產流轉關係。有了口徑和數據血緣關係的描述，就極大降低了數據解釋成本。
基於之上的整個生產數據流，再來看下ETL內部是如何實現流轉的。在前面提到我們有基於事實明細聚合查詢後回填到事實寬表的查詢，比如在訂單維度，標記該訂單是否為用戶的首個完成單。同時我們有接入api推送的數據，這類事實數據入庫後，在下游應用在一些實時的dashboard上。還有針對遲到的事實這種場景，需要有單個字段回溯的場景，舉個例子，比如訂單狀態，正常的餐飲訂單一般在幾個小時候才完成，但是會有些預訂單或物流單，在訂單創建數週之後才能完成。那麼基於以上場景，需要有一套支持細粒度更新的數據存儲解決方案，這裡就引入了一層ODS，所有的數據拼裝和轉換結果都先入到ODS中，再sqoop到集群，同時也會有從集群聚合查詢後回填到ods的場景。
我們對ODS的定位：第一，在業務系統和倉庫之間形成一個隔離層，ODS用於存放從業務系統直接抽取出來的數據，這些數據從數據結構、數據之間的邏輯關係上都與業務系統基本保持一致，因此在抽取過程中極大降低了數據轉化的複雜性，而主要關注數據抽取的接口、數據量大小、抽取方式等方面的問題。第二，承擔一部分業務系統的明細查詢。對於明細查詢的場景一般有三種，帶主鍵的檢索式查詢、有時效性要求的OLTP條件檢索、離線場景的明細拉取或聚合，對於前兩種場景是適合放在ODS中的，面向OLAP的查詢引擎無法滿足其時效性。第三，支持業務的回溯更新場景。
在ODS的選型上，考慮過三種類型的存儲：行增量式、列式、支持OLTP+OLAP混合的存儲。這三種類型的存儲，對應到這個數據場景下，需要不同的解決方案。行增量式以druid為例來講，高性能寫入，不支持更新。在外賣這邊druid已經應用於日誌的一些場景，那麼能不能用它來解決業務數據呢，在一些特定場景下是可以的，外賣的絕大多數業務都是有主鍵屬性的，還是拿訂單狀態來說，把訂單的每次變更都當作一個事件，記錄在druid中，下游的ETL按事件發生順序倒敘排列取最新狀態即可，但帶來的問題就是會冗餘存儲大量過程狀態，數據量級膨脹的係數取決於變化維度的枚舉值個數。列式存儲，基於MPP架構的以vertica和greenplum為例，其在大數據場景下的查詢能力非常出色，但是針對數據回溯的場景，其滿足不了批量DML操作的低延時性能要求。
我們需要的是一個融合低延遲寫入和高性能分析的存儲系統，結構化存儲系統在hadoop生態裡通常分為兩類：靜態數據和動態數據。靜態數據，數據通常都是使用二進制格式存放到 HDFS 上面，譬如 Apache Avro，Apache Parquet。但無論是 HDFS 還是相關的系統，都是為高吞吐連續訪問數據這些場景設計的，都沒有很好的支持單獨 record 的更新，或者是提供好的隨機訪問的能力。 動態數據，數據通常都是使用半結構化的方式存儲，譬如 Apache HBase，Apache Cassandra。這些系統都能低延遲的讀寫單獨的 record，但是對於一些像 SQL 分析這樣需要連續大量讀取數據的場景，顯得有點捉緊見拙。上面的兩種系統，各有自己的側重點，一類是低延遲的隨機訪問特定數據，而另一類就是高吞吐的分析大量數據。簡單來講，我們需要的是一個OLTP+OLAP融合的存儲系統，kudu和phoenix就是在兩個特性之間找到了一個平衡點。我們線上目前用的是phoenix，它是一個基於hbase之上的組件，通過cm安裝部署非常簡易，因為是基於hbase的所以寫入性能可擴展，而且hbase的運維比較成熟，作為一個新型引擎引入技術風險較小。Kudu作為當下的一個主流趨勢，我們也搭建了測試集群正在摸索中。
這裡列了兩張phoenix官方提供的OLAP場景下和hive、impala的性能對比圖，在行數據遞增的時候其查詢性能耗時增長緩慢。在我們的實際應用經驗中，phoenix的性能主要得益於內存的使用，當DML操作比較密集的時候同時進行復雜的DQL操作，內存吃緊的狀態下其性能會顯著下降，所以在外賣的使用場景中，phoenix主要用於解決行粒度更新的問題，複雜的OLAP操作在impala和hive中。
前面提到了調度系統由任務(job)->轉換(transformer)->算子(operator)構成，算子是最小執行單元，在ETL的作業執行中，具體的一個算子就對應一個ETL作業，調度事先將作業在線上集群初始化，通過時間觸發+任務依賴的方式選擇宿主機調起任務，在宿主機選擇時優先選擇負載最低的節點執行。任務和轉換在同一層級中，依賴關係均是有向無環圖，每個節點的父子節點可能都會有多個。算子作為最小執行單元，一個轉換中可能會有多個算子。在調度系統中為ETL設置了多種算子類型，比如ETL算子、sqoop算子、webservice算子等，通過預定義的參數填寫來配置。
下面來講講ETL框架的流程設計。框架最核心的設計理念，是從ETL的場景出發，可以分為兩大類。這兩個類別能覆蓋大約90%以上的需求，一些特殊場景基於配置無法滿足的，按照框架定義的標準進行編碼自定義class實現（比如數據源是按業務id hash分表，比如數據主體沒有業務主鍵）。第一類，前面提到的字段之間多對多的映射關係，從上游查詢出多個字段，經過轉換後插入下游某個表，如果轉換邏輯可通過sql實現，對應的其實就是一個insert from select 的語法。那麼在異構存儲的系統中，可能會涉及到多個數據源的相互轉換，無法用一個sql搞定，這個就是開放式sql做的事情，將這類的生產邏輯都轉化為一張配置表，將數據抽取和裝載分開執行。第一類是相對簡單且常見的場景，那麼針對轉換邏輯無法通過sql實現的場景，我們設計了開放式index，一個針對字段粒度的etl方式。圖中開發式index的配置裡db¬_connect、db_connect_table、select_sql三個參數描述從哪去取源數據，db_connect_engine、fact_table、index_column描述數據往哪插入，寫入哪個字段，對於需要用coding進行邏輯轉換的，在handle_flag中指定自定義函數，自定義函數通過約束統一輸入輸出的格式來對接到上下游數據流，剩下的字段多為配置的描述性字段，如是否可回溯、指標分類、生效時間等。
基於這兩類配置來看下流程圖，首先初始化參數和配置，比如要生產表明為t1的表，根據參數中的設定檢查對應目標表的分區是否存在，不存在就先創建，然後寫入信號燈，告訴信號燈t1表開始生產了。開始進入正式的生產流程，第一步判斷是否有自定義編碼，有的話先加載執行，第二步查詢open sql的配置，如果發現t1表的生產對應多條開放式sql的配置，則順序執行，同時也支持在傳參入口傳入指定配置編號來細粒度執行。第三步查詢open index，得到多條配置，以計算t1.f1 和t1.f2為例，兩個字段對應不同的數據源不同的邏輯，先獨立執行數據抽取，然後執行各自獨立的自定義函數（如果有的話），然後再對處理結果進行merge，拼裝成一個upsert語句進行插入。第四步，調用監控服務判斷生產結果是否符合預期標準。第五步，寫入信號燈告訴信號燈生產完成。整體ETL的生產，結合了ODS支持upsert的特性，加上open index的細粒度生產配置，實現了數據的細粒度生產與回溯；生產作業命令的參數支持按表和多字段傳入，實現了計算粒度可伸縮，可整表批處理生產也可生產單個字段；同時支持ETL配置熱加載，隨時修改隨時上線。
這是一個開放式sql的配置示例，截圖了配置的部分字段，可以看到前兩個字段描述了數據源，然後描述了數據往哪寫，在轉換邏輯中，查詢sql的字段和寫入sql的字段是一一映射的。
是一個open index中函數編碼的示例，計算“代理商抽佣金額”這個指標，函數的入口統一標準格式，輸入為主題數據的二維數組，數組的key為業務主鍵，數組第一維表示數據行，第二維表示數據列，在這個示例中需要對入參的多列分別json反解析後再進行數學邏輯運算，結果集作為新增列再填回維護數組並返回。針對這種業務邏輯複雜，需要反解自定義加密、解析複雜的數據結構的場景，我們只能編寫業務代碼來處理，open index支持自定義函數的設計很好的解決了這方面的擴展性。
除了ETL計算的流程之外，整個ETL系統中還有重要的一環就是sqoop，我們對開源工具進行了封裝，在其基礎上加入了分區自動創建、併發控制、字段類型映射等功能。現在來看下sqoop的處理流程，首先在選擇數據源、創建分區後，第一步先進行schema diff，做這一步主要是為了解決線上庫schema變更下游未及時更改導致sqoop失敗的問題，且線上業務變更數據庫的操作經常在夜間進行，對大數據運維來講同步跟進的成本很高，且夜間是數據生產的高峰期。這裡我們開發了一套工具用來事前錄入變更schema，通過版本控制，在檢測到schema有差異時自動去加載最新的schema版本對線上hive/impala進行變更。第二步針對要同步的數據count量級，一方面是用於併發的邊界處理，一方面用於同步完成後進行數據量級校驗。第三步進行邊界處理，即對應原生工具中的-split-by和-m參數，用於控制同步的併發度和數據傾斜。第四步進行字段映射處理，由於sqoop本質是通過一個mr任務抽取數據後生成數據文件，hive在建表關聯時會存在很多字段映射的問題，比如mysql中的bigint unsigned 類型，但在hive中只有bigint類型，符號位佔據了一位所以其長度不夠，這種情況都需要在hive中轉為string。第五步是基於配置的壓縮處理，對應—compress參數。以上幾步完成sqoop的命令拼裝，執行後就產出數據文件了，之後釋放數據生產完成信號，然後針對同步的表進行hdfs緩存池的維護，最後在impala進行元數據的compute（數據出口有hive、impala、greenplum三種引擎）。
以上，介紹了數據生產的核心流程，但從系統的層面來講，數據生產完成不等於數據可用。從數據生產到交付至業務使用的場景中，經常會遇到數據生產任務執行異常、生產任務延遲、生產結果異常等，對於任務執行異常或延時，調度系統都會進行重試和報警，對於生產結果的異常，就得從數據本身去判斷處理了。
對於數據的質量問題，我們在生產任務的前後都設置了checkpoint，前置的checkpoint用於檢查它所依賴的上游數據是否符合預期，後置的checkpoint用於檢查自身產出的數據是否符合標準。那麼具體的邏輯就是由監控服務來實現，和常見的監控服務類似，由用戶自定義配置監控項，進行定時check。在數據監控服務中，和常規系統監控服務的差異在於這裡的配置分為表結構監控和數據可用性監控，後者由sql實現同源或異源數據的對比、環比同比的閾值監控、空值率及類型監控，同時對於不同閾值觸發不同的監控報警級別和處理措施，例如“訂單量”環比波動超過20%就郵件報警，超過50%就短信電話報警，“流量轉化率”為負數就停止生產任務等。監控服務同時提供api觸發check的功能，由調度的webservice算子或執行腳本進行自主調用，更加靈活的支持在關鍵流程中設置check任務，檢測到異常則終止整個生產任務流，下游生產將不會被觸發直到異常點恢復。
對於數據交付流程上來講，用戶還會關心任務的生產進度，信號燈服務從命名上就可以看出是用來釋放數據生產信號的。在前面介紹ETL作業流程中就已經提到，在生產初始和結束時都會調用信號燈的接口，來釋放信號。信號燈服務提供了信號寫入api和信號查詢api，對於下游系統來講只需要監測數據可用信號即可，在服務層面實現了通過api解耦。在數據的上下游關係中，業內常見的依賴都是整表依賴，整表依賴的最大痛點在於對於寬表來說，下游的生產開始時間取決於上游最後一個字段的生產結束時間，而依賴關係中可能只是使用到了上游的部分字段。在前面介紹的ETL生產流程中，我們的ETL系統已經支持到了字段粒度的生產，那麼對於數據交付來講，信號燈的信號釋放表現在字段粒度，就可以做到字段粒度的作業流依賴。
以上，構建了完整的數據交付體系
在數據使用層面，數據的可解釋性是所有大數據團隊的一個痛點。數據可解釋主要包含兩個方面：數據口徑和數據依賴關係。ETL開發者經常面臨的一個問題就是要向數據使用方解釋你的數據是如何生產出來的。在業內常見的生產依賴關係都是數據表和生產任務之間的依賴，對於生產任務具體用到了上游數據表中的哪些字段只有在編碼邏輯中才能體現，是無法暴露給數據使用方的。由於我們的ETL系統絕大多數生產邏輯都是基於配置的，因此通過配置文件即可解析出數據上下游的生產依賴關係 - -即數據血緣關係。在調度配置的任務中，一個任務對應一張表的生產，通過血緣關係可以清晰看到這個任務在數據層面的依賴關係，做到數據可解釋、生產過程可解釋。
以上，是今天分享的全部內容。
相關推薦
