Apache Flink在唯品會的實踐
唯品會實時平臺現狀
目前在唯品會實時平臺並不是一個統一的計算框架,而是包括Storm,Spark,Flink在內的三個主要計算框架。由於歷史原因,當前在Storm平臺上的job數量是最多的,但是從去年開始,業務重心逐漸切換到Flink上面,所以今年在Flink上面的應用數量有了大幅增加。
實時平臺的核心業務包含八大部分:實時推薦作為電商的重點業務,包含多個實時特徵;大促看板,包含各種維度的統計指標(例如:各種維度的訂單、UV、轉化率、漏斗等),供領導層、運營、產品決策使用;實時資料清洗,從使用者埋點收集來資料,進行實時清洗和關聯,為下游的各個業務提供更好的資料;此外還有網際網路金融、安全風控、與友商比價等業務,以及Logview、Mercury、Titan作為內部服務的監控系統、VDRC實時資料同步系統等。
實時平臺的職責主要包括實時計算平臺和實時基礎資料。實時計算平臺在Storm、Spark、Flink等計算框架的基礎上,為監控、穩定性提供了保障,為業務開發提供了資料的輸入與輸出。實時基礎資料包含對上游埋點的定義和規範化,對使用者行為資料、SQL/">MySQL的Binlog日誌等資料進行清洗、打寬等處理,為下游提供質量保證的資料。
在架構設計上,包括兩大資料來源。一種是在App、微信、H5等應用上的埋點資料,原始資料收集後傳送到在kafka中;另一種是線上實時資料的MySQL Binlog日誌。資料在計算框架裡面做清洗關聯,把原始的資料通過實時ETL為下游的業務應用(包括離線寬表等)提供更易於使用的資料。
Flink在唯品會的實踐
場景一:Dataeye實時看板
Dataeye實時看板是支援需要對所有的埋點資料、訂單資料等進行實時計算時,具有資料量大的特點,並且需要統計的維度有很多,例如全站、二級平臺、部類、檔期、人群、活動、時間維度等,提高了計算的複雜程度,統計的資料輸出指標每秒鐘可以達到幾十萬。
以UV計算為例,首先對Kafka內的埋點資料進行清洗,然後與Redis資料進行關聯,關聯好的資料寫入Kafka中;後續Flink計算任務消費Kafka的關聯資料。通常任務的計算結果的量也很大(由於計算維度和指標特別多,可以達到上千萬),資料輸出通過也是通過Kafka作為緩衝,最終使用同步任務同步到HBase中,作為實時資料展示。同步任務會對寫入HBase的資料限流和同型別的指標合併,保護HBase。與此同時還有另一路計算方案作為容災。
在以Storm進行計算引擎中進行計算時,需要使用Redis作為中間狀態的儲存,而切換到Flink後,Flink自身具備狀態儲存,節省了儲存空間;由於不需要訪問Redis,也提升了效能,整體資源消耗降低到了原來的1/3。
在將計算任務從Storm逐步遷移到Flink的過程中,對兩路方案先後進行遷移,同時將計算任務和同步任務分離,緩解了資料寫入HBase的壓力。
切換到Flink後也需要對一些問題進行追蹤和改進。對於FlinkKafkaConsumer,由於業務原因對kafka中的Aotu Commit進行修改,以及對offset的設定,需要自己實現支援kafka叢集切換的功能。對不帶window的state資料需要手動清理。還有計算框架的通病——資料傾斜問題需要處理。同時對於同步任務追數問題,Storm可以從Redis中取值,Flink只能等待。
場景二:Kafka資料落地HDFS
之前都是通過Spark Streaming的方式去實現,現在正在逐步切換到Flink上面,通過OrcBucketingTableSink將埋點資料落地到HDFS上的Hive表中。在Flink處理中單Task Write可達到3.5K/s左右,使用Flink後資源消耗降低了90%,同時將延遲30s降低到了3s以內。目前還在做Flink對Spark Bucket Table的支援。
場景三:實時的ETL
對於ETL處理工作而言,存在的一個痛點就是字典表儲存在HDFS中,並且是不斷變化的,而實時的資料流需要與字典表進行join。字典表的變化是由離線批處理任務引起的,目前的做法是使用ContinuousFileMonitoringFunction和ContinuousFileReaderOperator定時監聽HDFS資料變化,不斷地將新資料刷入,使用最新的資料去做join實時資料。
我們計劃做更加通用的方式,去支援Hive表和Stream的join,實現Hive表資料變化之後,資料自動推送的效果。
Flink On K8S
在唯品會內部有一些不同的計算框架,有實時計算的,有機器學習的,還有離線計算的,所以需要一個統一的底層框架來進行管理,因此將Flink遷移到了K8S上。
在K8S上使用了思科的網路元件,每個docker容器都有獨立的ip,對外也是可見的。實時平臺的融合器整體架構如下圖所示。
唯品會在K8S上的實現方案與Flink社群提供的方案差異還是很大的。唯品會使用K8S StatefulSet模式部署,內部實現了cluster相關的一些介面。一個job對應一個mini cluster,並且支援HA。對於Flink來說,使用StatefulSet的最大的原因是pod的hostname是有序的;這樣潛在的好處有:
hostname為-0和-1的pod可以直接指定為jobmanager;可以使用一個statefulset啟動一個cluster,而deployment必須2個;Jobmanager和TaskManager分別獨立的deployment。
pod由於各種原因fail後,由於StatefulSet重新拉起的pod的hostname不變,叢集recover的速度理論上可以比deployment更快(deployment每次主機名隨機)。 映象的docker entrypoint腳本里面需要設定的環境變數設定說明:
| 環境變數名稱 | 引數 | 示例內容 | 說明 | |--- |---|---|---|---| | JOB_MANGER_HOSTS | StatefulSet.name-0,StatefulSet.name-1 | flink-cluster-0,flink-cluster-1 | JM的主機名,短主機名;可以不用FQDN | | FLINK_CLUSTER_IDENT | namespace/StatefulSet.name | default/flink-cluster | 用來做zk ha設定和hdfs checkpiont的根目錄 | | TASK_MANAGER_NUMBER_OF_TASK_SLOTS | containers.resources.cpu.limits | 2 | TM的slot數量,根據resources.cpu.limits來設定 | | FLINK_ZK_QUORUM | env:FLINK_ZK_QUORUM | 10.198.199.112:2181 | HA ZK的地址 | | JOB_MANAGER_HEAP_MB | env:JOB_MANAGER_HEAP_MB value:containers.resources.memory.limit -1024 | 4096 | JM的Heap大小,由於存在堆外記憶體,需要小於container.resources.memory.limits;否則容易OOM kill | | TASK_MANAGER_HEAP_MB | env:TASK_MANAGER_HEAP_MB value: containers.resources.memory.limit -1024 |4096 | JM的Heap大小,由於存在堆外記憶體,需要小於container.resources.memory.limits;否則容易OOM kill |
對應Flink叢集所依賴的HDFS等其他配置,則通過建立configmap來管理和維護。
kubectl create configmap hdfs-conf --from-file=hdfs-site.xml --from-file=core-site.xml
後續計劃
當前實時系統,機器學習平臺要處理的資料分佈在各種資料儲存元件中,如Kafka、Redis、Tair和HDFS等,如何方便高效的訪問,處理,共享這些資料是一個很大的挑戰,對於當前的資料訪問和解析常常需要耗費很多的精力,主要的痛點包括:
- 對於Kafka,Redis,Tair中的binary(PB/Avro等格式)資料,使用者無法快速直接的瞭解資料的schema與資料內容,採集資料內容及與寫入者的溝通成本很高。
- 由於缺少獨立的統一資料系統服務,對Kafka,Redis,Tair等中的binary資料訪問需要依賴寫入者提供的資訊,如proto生成類,資料格式wiki定義等,維護成本高,容易出錯。
- 缺乏relational schema使得使用者無法直接基於更高效易用的SQL或LINQ層API開發業務。
- 無法通過一個獨立的服務方便的釋出和共享資料。
- 實時資料無法直接提供給Batch SQL引擎使用。
- 此外,對於當前大部分的資料來源的訪問也缺少審計,許可權管理,訪問監控,跟蹤等特性。
UDM(統一資料管理系統)包括Location Manager, Schema Metastore以及Client Proxy等模組,主要的功能包括:
- 提供從名字到地址的對映服務,使用者通過抽象名字而不是具體地址訪問資料。
- 使用者可以方便的通過Web GUI介面方便的檢視資料Schema,探查資料內容。
- 提供支援審計,監控,溯源等附加功能的Client API Proxy。
- 在Spark/Flink/Storm等框架中,以最適合使用的形式提供這些資料來源的封裝。
UDM的整體架構如下圖所示:
UDM的使用者包括實時,機器學習以及離線平臺中資料的生產者和使用者。在使用Sql API或Table API的時候,首先完成Schema的註冊,之後使用Sql進行開發,降低了開發程式碼量。
在Flink中,使用UDMExternalCatalog來打通Flink計算框架和UDM之間的橋樑,通過實現ExternalCatalog的各個介面,以及實現各自資料來源的TableSourceFactory,完成Schema和接入管控等各項功能。
關於作者:王新春目前在唯品會負責實時平臺相關內容,主要包括實時計算框架和提供實時基礎資料,以及機器學習平臺的工作。之前在美團點評,也是負責大資料平臺工作。他已經在大資料實時處理方向積累了豐富的工作經驗。