深度介紹Flink在字節跳動數據流的實踐

本文是字節跳動數據平台開發套件團隊在1月9日Flink Forward Asia 2021: Flink Forward 峰會上的演講分享，將着重分享Flink在字節跳動數據流的實踐。

字節跳動數據流的業務背景

數據流處理的主要是埋點日志。埋點，也叫Event Tracking，是數據和業務之間的橋梁，是數據分析、推薦、運營的基石。

用戶在使用App、小程序、Web等各種線上應用時產生的行為，主要通過埋點的形式進行采集上報，按不同的來源分為客戶端埋點、Web端埋點、服務端埋點。

不同來源的埋點都通過數據流的日志采集服務接收到MQ，然后經過一系列的Flink實時ETL對埋點進行數據標准化、數據清洗、實時風控反作弊等處理，最終分發到下游，主要的下游包括ABTest、推薦、行為分析系統、實時數倉、離線數倉。

所以，如果用一句話來概括數據流主要業務，其實就是埋點的收集、清洗、分發。

目前在字節跳動，清洗和分發環節是基於Flink搭建的。

01 - 數據流業務規模

• 業務數量：在 字節跳動，包括抖音、今日頭條、西瓜視頻、番茄小說在內的3000多個大大小小的APP和服務都接入了數據流。
• 數據流峰值流量：當前，字節跳動埋點數據流峰值流量超過1億每秒，每天處理超過萬億量級埋點，PB級數據存儲增量。
• ETL任務規模：目前，字節跳動數據流在多個機房部署超過1000個Flink任務和超過1000個MQ Topic，使用超過50W Core CPU，單任務最大12W Core CPU ，Topic最大10000 Partitio。

02 - 數據流業務挑戰

字節跳動數據流ETL遇到的挑戰主要有四點：

• 第一點，流量大，任務規模大。

• 第二點，處在所有產品數據鏈路最上游，下游業務多，ETL需求變化頻繁。

• 第三點，高SLA要求，下游推薦、實時數倉等業務對穩定性和時效性有比較高的要求。

• 最后一點，在流量大、業務多、SLA要求高的情況下，針對流量、成本、SLA保障等多維度的綜合治理也面臨挑戰。

下面從兩個數據流業務場景中介紹一下我們遇到的業務挑戰。

1、UserAction ETL場景

在UserAction ETL場景中，我們遇到的核心需求是：種類繁多且流量巨大的客戶端埋點需求和ETL規則動態更新的需求。

在字節內部，客戶端的埋點種類繁多且流量巨大，而推薦關注的只是部分埋點，因此為了提升下游推薦系統處理效率，會在數據流配置一些ETL規則，對埋點進行過濾，並對字段進行刪減、映射、標准化之類的清洗處理，將埋點打上不同的動作類型標識。

處理之后的埋點一般稱之為UserAction，UserAction數據會和服務端展現等數據在推薦Joiner任務的分鍾級窗口中進行拼接Join，產出Instance訓練樣本。

舉個例子：一個客戶端的文章點贊埋點描述了用戶在一個時間點對某一篇文章進行了點贊操作，埋點經過數據流日志采集服務進入數據流ETL鏈路，通過UserAction ETL處理后實時地進入到推薦Joiner任務中拼接生成樣本更新推薦模型，從而提升用戶體驗。

如果產出UserAction數據的ETL鏈路出現比較大的延遲，那么就不能在窗口內及時完成拼接，可能導致用戶體驗下降。

因此對於推薦來說，數據流的時效性是一個強需求。

而推薦模型的迭代、產品埋點的變動都可能導致UserAction的ETL規則的變動。如果ETL規則硬編碼在代碼中，每次修改都需要升級代碼並重啟Flink Job，會影響數據流穩定性和數據的時效性。因此，這個場景的另一個需求就是ETL規則的動態更新。

2、數據分流場景

目前，抖音業務的埋點Topic晚高峰流量超過1億/秒，而下游電商、直播、短視頻等不同業務的實時數倉關注的埋點范圍實際上都只是其中的一小部分。

如果各業務分別使用一個Flink任務，消費抖音埋點Topic，過濾消費各自關注的埋點，需要消耗大量Yarn資源，同時會造成MQ集群帶寬扇出嚴重，影響MQ集群的穩定性。

因此，數據流提供了數據分流服務，使用一個Flink任務消費上游埋點Topic，然后通過配置規則的方式，將各業務關注的埋點分流到下游小Topic中，再提供給各個業務消費。這樣就減少了不必要的反序列化開銷，同時降低了MQ集群帶寬扇出比例。

在數據分流場景中，核心需要解決的是高穩定的SLA。因為斷流、數據延遲可能會影響推薦效果、廣告收入、實時數據報表。

同時隨着業務發展，實時數據需求日益增加，分流規則新增和修改也會日益頻繁。如果每次規則變動都需要修改代碼並重啟Flink Job，會影響很多下游，因此分流規則的動態更新也是這一場景中的強需求。

字節跳動數據流實踐

01-數據流ETL鏈路建設

字節跳動數據流ETL鏈路建設主要經歷了三個階段：

第一階段是2018年以前業務需求快速迭代的早期階段

主要使用PyJStorm和基於Python的規則引擎構建主要的流式數據處理鏈路。其特點是比較靈活，可以快速支持業務需求。

但隨着埋點流量快速上漲，PyJStorm暴露出很多穩定性和運維上的問題，性能也不足以支撐業務的增長。

2018年，公司內部開始大力推廣Flink，並且針對大量舊任務使用PyJStorm的情況，提供了PyJStorm到PyFlink的兼容適配。流式任務托管平台的建設一定程度上解決了流式任務運維管理的問題。數據流ETL鏈路也在2018年全面遷移到了PyFlink，進入了流式計算的新時代。

第二個階段是2018至2020年

隨着流量的進一步上漲，PyFlink和Kafka的性能瓶頸、以及JSON數據格式帶來的性能和數據質量問題都一一顯現出來，與此同時下游業務對延遲、數據質量的敏感程度卻是與日俱增。

於是，我們一方面對一些痛點進行了針對性的優化。另一方面，花費1年多的時間將整個ETL鏈路從PyFlink切換到了Java Flink，使用基於Groovy的規則引擎替換了基於Python的規則引擎，使用ProtoBuf替換了JSON。

數據流ETL新鏈路，相比舊鏈路性能提升了1倍。

與此同時，一站式大數據開發平台和流量平台的建設提升了數據流在任務開發運維、ETL規則管理、埋點元數據管理、多機房容災降級等多方面的能力。

第三個階段是從2021年開始

在全球資源供應緊張的背景下，進一步提升數據流ETL性能和穩定性，滿足流量增長和需求增長的同時，降低資源成本和運維成本，是這一階段的主要目標。我們主要從三個方面進行了優化：

• 優化引擎性能。隨着流量和ETL規則的不斷增加，基於Groovy的規則引擎使用的資源也不斷增加，於是我們基於Janino進行了重構，引擎性能得到數倍提升。

• 優化埋點治理體系。我們基於流量平台建設了一套比較完善的埋點治理體系，通過無用埋點下線、埋點采樣等手段降低埋點成本。

• 優化鏈路。我們進行了鏈路分級，不同等級的鏈路保障不同的SLA，在資源不足的情況下優先保障高優埋點鏈路。

從2018年到2020年，我們持續在數據流Flink ETL Job應對需求挑戰上取得了一些實踐效果。下圖展示了數據流Flink ETL Job是如何支持動態更新的，在不重啟任務的情況下，實時更新上下游Schema、規則處理邏輯、修改路由拓撲。

流量平台Config Center為數據流Flink ETL Job提供上下游數據集拓撲關系、Schema、ETL規則和UDF等元數據。

數據流Flink ETL Job中的每個TaskManager中會有一個Meta Updater更新線程，更新線程每分鍾通過RPC請求從流量平台拉取並更新相關元數據。

Source將從MQ中消費到的數據傳入ProcessFunction，根據MQ對應的Schema反序列化為InputMessage，然后進入規則引擎中，通過規則索引匹配出需要運行的規則，每條規則抽象為一個Filter模塊和一個action模塊，Filter和action都支持UDF ，Filter篩選命中后，通過action模塊對輸入數據進行字段映射和清洗，然后寫出到OutputMessage中。

每條規則也指定了對應的下游數據集，路由信息也會一並寫出到OutputMessage。OutputMessage輸出到Sink后，Sink根據OutputMessage中的路由信息將數據發送到SinkManager管理的Client，由對應的Client發送到下游MQ。

這里解釋一下我們為什么讓每個TaskManager通過一個MetaData updater定時去更新元數據，而不是通過增加一條元數據流來更新。這么做的原因主要是因為使用元數據流更新的方式需要開啟Checkpoint以保存元數據的狀態，而在字節跳動數據流這樣的大流量場景下，開啟Checkpoint會導致在Failover時產生大量重復數據，下游無法接受。

1、規則引擎的解決方案

數據流Flink ETL Job使用的規則引擎經歷了從Python到Groovy再到Janino的迭代。規則引擎對於數據流來說最主要的就是提供動態更新ETL規則的能力。

Python由於腳本語言本身的靈活性，動態加載規則實現起來比較簡單，通過Compile函數可以將一段規則代碼片段編譯成字節代碼，再通過eval函數進行調用即可。但存在性能較低，規則缺乏管理的問題。

遷移到Java Flink后，我們在流量平台上統一管理ETL規則、Schema、數據集等元數據。用戶在流量平台編輯ETL規則，規則從前端視圖發送到后端，經過一系列校驗后保存為邏輯規則，引擎將邏輯規則編譯為物理規則運行。Groovy本身兼容Java，所以我們可以通過GroovyClassLoader動態的加載規則、UDF。

但使用Groovy，雖然性能比Python提高了很多倍，但額外的開銷仍比較大，因此我們又借助Janino可以高效動態編譯Java類並加載到JVM直接執行的能力，將Groovy替換為Janino。

除了規則引擎的迭代，我們在平台側的測試、發布、監控和報警方面也做了很多建設。

測試發布環節支持了規則的線下測試、線上調試、灰度發布等功能，監控環節則是支持字段、規則、任務等不同粒度的異常監控，並支持了規則流量的波動報警、任務的資源報警等功能。

規則引擎的應用解決了數據流ETL鏈路如何快速響應業務需求的問題，實現了動態調整ETL規則不需要修改代碼、重啟任務。

但規則引擎本身的迭代、流量增長導致的資源擴容等場景還是需要升級重啟Flink任務，引發斷流。除了重啟斷流外，大任務還可能遇到啟動慢、隊列資源不足或資源碎片導致起不來等問題。

2、Flink拆分任務的實踐

針對這些痛點，我們上線了Flink拆分任務。Flink拆分任務本質上就是將一個大任務拆分為一組子任務，每個子任務按比例消費上游Topic一部分Partition，處理后再分別寫出到下游Topic。

舉個例子，上游Topic有200個Partition，我們在大數據研發治理套件DataLeap的數據開發上配置一個Flink拆分任務只需要指定每個子任務的流量比例，其余參數都可以按比例自動同步。

拆分任務的應用使得數據流Flink ETL Job除了規則粒度的灰度發布能力，還具備了Job粒度的灰度發布能力，從此升級、擴容不斷流，上線風險更可控。同時，由於拆分任務各子任務是獨立的，因此單個子任務出現反壓、fail-over不會影響其他子任務，對下游的影響更小。另外一個優點是單個子任務資源使用量更小，子任務可以同時在多個隊列靈活部署。

在流量迅速增長的階段，數據流最開始是通過Kafka Connector直接寫Kafka。但是由於數據流Flink ETL Job任務處理的流量大，Sink比較多，批量發送的效率不高，Kafka集群寫入請求量很大，另外由於每個Sink一個Client，Client與Kafka集群間建立的連接數很多，而Kafka集群由於Controller性能瓶頸也無法繼續擴容。

為了緩解Kafka集群壓力，數據流Flink ETL Job引入了DataBus組件。

DataBus以Agent的方式部署Yarn節點上，Agent中每個Channel對應一個Kafka Topic。數據流FlinkETL Job每個TM中的SinkManager使用DataBus Client 通過 Unix Domain Socket的方式將數據發送到DataBus Agent 的Channel中，再由Channel將數據批量發送到對應的Kafka Topic。

由於每個Yarn節點上所有的TM都先把數據發送到本機的Databus Agent，每個Databus channel聚合了機器上所有TM Sink寫同一個Topic的數據，因此批量發送的效率非常高，極大的降低了Kafka集群的寫入請求量，與Kafka集群之間需要建立的連接也更少。

同時，單個請求中數據條數的增加帶來更高的壓縮效率，在Databus  Agent 上開啟了ZSTD壓縮后，Kafka集群寫入帶寬降低了37%，極大的緩解了Kafka集群的壓力。

春晚活動是萬眾矚目的一大盛事，2021年春晚活動期間數據流對相關的埋點鏈路進行了重點保障。

首先是完成了多機房的容災部署並准備了多種切流預案，正常情況下流量會均勻的打到多個機房，MQ多機房同步，Flink ETL Job都從本地消費。如果某個機房出現網絡或其他大規模故障，可以從客戶端將流量調度到其他機房，也可以在CDN側將流量調度到不同的機房，數據流Flink ETL 鏈路可以分鍾級進入容災模式，切換到可用機房。

為了應對口播期間的流量洪峰，我們還准備了客戶端降級策略與服務端降級策略。其中客戶端降級策略可以動態的降低一定百分比用戶的埋點上報頻率，口播期間不上報，口播結束后逐步恢復。

在降級場景下，下游指標計算可以通過消費未降級的活動埋點分流估算整體指標。春節活動鏈路的順利保障標志着數據流基於Flink搭建的ETL鏈路已經能提供較好的穩定性和可用性。

02 - 數據流治理實踐

數據流比較常見的治理問題包括但不限於以下幾個：

• 第一個是數據流穩定性治理中最常見的一個問題——Yarn單機問題導致Flink任務fail-over、反壓、消費能力下降。Yarn單機問題的類型有很多，比如：隊列負載不均、單機load高、其他進程導致CPU負載高、硬件故障等等。

• 第二個問題是Kafka集群負載不均導致Flink任務生產消費受到影響。

• 第三個問題是埋點治理場景中無用埋點、異常埋點消耗大量計算存儲資源。

• 針對單機問題，我們從Flink和Yarn兩個層面分別進行了優化，將單機load高導致的延遲減少了80%以上。

首先，Flink層面的優化。

在數據流ETL場景中，為了減少不必要的網絡傳輸，Partitioner主要采用Rescale Partitioner。而Rescale Partitioner會使用Round-robin的方式發送數據到下游部分Channel中，由於單機問題可能出現個別任務處理能力不足的情況，導致反壓，任務出現lag。

實際上數據發到下游任何一個任務都是可以的，最合理的策略應該根據下游任務的處理能力去發送數據。

另一方面，我們注意到Flink Credit-based Flow Control反壓機制中，可以通過Backlog Size判斷下游任務的處理負載，那么我們就可以將Round-robin發送的方式修改為根據Channel的Backlog Size信息選擇負載更低的下游Channel發送的方式。

方案上線后隊列的負載更加均衡，CPU利用率提升10%。

其次，Yarn層面的優化。

第一、隊列資源使用獨立Label隊列，避免高峰期和其他低優任務互相影響；

第二、Yarn節點上的DataNode偶發有帶寬打滿、CPU使用高的情況，影響節點上數據流Flink ETL 任務的穩定性，通過給DataNode設置網絡限速並進行CPU綁核以避免DataNode對Flink進程的影響；

第三、Yarn反調度策略。目前字節跳動Flink使用的Yarn GangScheduler調度策略會根據約束條件選擇性的獲取分配到的Yarn資源，在任務啟動時做到比較均衡的放置Container，但由於時間的推移，流量的變化等諸多因素，隊列還是可能會出現負載不均衡的情況。

反調度策略則是為了解決負載不均衡而生的二次調度機制。Yarn會定期檢查集群中不再滿足原有約束的Container，並在這些Container所在的節點上篩選出需要重新調度的Container返回給Flink JobManager，Flink會重新調度這些Container。

重新調度會按照原有約束嘗試申請等量的可用資源，申請成功后進行遷移，申請不成功不做操作。

針對Kafka集群優化問題，我們自研來了存儲計算分離的MQ——BMQ，單GB流量成本下降50%。

在數據流這種大流量場景下使用Kafka，經常會遇到broker或者磁盤負載不均衡、磁盤壞掉等情況，進行擴容、機器替換時的運維操作會引起集群Under Replica, 影響讀寫性能。除此之外，Kafka還有集群規模瓶頸、多機房容災部署成本高等缺點。

為了優化這些問題，BMQ這款字節跳動自研的存儲計算分離的MQ應運而生。

BMQ數據使用HDFS分布式存儲，每個partition被切分為多個segment，每個segment對應一個HDFS文件，元數據使用kv存儲，Proxy和Broker都是無狀態的，因此可以支持快速擴縮容，且沒有數據拷貝不會影響讀寫性能。受益於HDFS多機房容災部署能力，BMQ多機房容災部署變得比較簡單，數據同時寫入所有容災機房成功后才會向client返回成功，數據消費則是在每個機房本地消費，減少了跨機房帶寬，除此之外，由於基於HDFS存儲所需的副本數更少，單GB流量成本下降50%。

針對埋點治理，我們從全產品開啟埋點管控、無用埋點監控&自助下線、埋點分級、風控能力建設四個點入手。

第一點，全產品開啟埋點管控。所有產品都需要先在流量平台注冊埋點元數據才能上報，這是從埋點接入流程進行的治理。

第二點，對於已上報的埋點，我們會通過埋點血緣，統計出已經沒有在使用的埋點，自動通知埋點負責人在平台進行自助下線。埋點注冊和埋點下線完成后，都會通過埋點管控服務動態下發相關的配置信息到埋點SDK和數據流Flink ETL任務中，從而保障未注冊埋點和無用埋點在上報或ETL環節被丟棄掉。

第三點是根據不同的用途對埋點進行分級，從而Dump到HDFS和數倉的時候可以按不同等級進行分區，不同等級的分區提供不同的TTL和就緒時間的保障。

最后一點則是針對異常流量，數據流ETL鏈路接入了風控系統，對埋點進行實時打標或過濾，防止異常流量造成數據傾斜、數據延遲、統計指標異常等問題。

目前，Flink在字節跳動數據流實踐中，已經可以做到計算層面的流批一體。接下來，我們還將計划探索計算和存儲的流批一體，同時也會探索雲原生架構，實現資源的動態Rescale，提升資源利用率。我們也會一些高優鏈路保障上追求更高的SLA，比如保障端到端Exactly-once語義。

目前，現有的能力已經通過火山引擎大數據研發治理套件DataLeap對外開放。

歡迎關注字節跳動數據平台同名公眾號