唯品會實時平臺現(xiàn)狀

目前在唯品會實時平臺并不是一個統(tǒng)一的計算框架，而是包括Storm，Spark，F(xiàn)link在內(nèi)的三個主要計算框架。由于歷史原因，當(dāng)前在Storm平臺上的job數(shù)量是最多的，但是從去年開始，業(yè)務(wù)重心逐漸切換到Flink上面，所以今年在Flink上面的應(yīng)用數(shù)量有了大幅增加。

實時平臺的核心業(yè)務(wù)包含八大部分：實時推薦作為電商的重點業(yè)務(wù)，包含多個實時特征;大促看板，包含各種維度的統(tǒng)計指標(biāo)(例如：各種維度的訂單、UV、轉(zhuǎn)化率、漏斗等)，供領(lǐng)導(dǎo)層、運營、產(chǎn)品決策使用;實時數(shù)據(jù)清洗，從用戶埋點收集來數(shù)據(jù)，進行實時清洗和關(guān)聯(lián)，為下游的各個業(yè)務(wù)提供更好的數(shù)據(jù);此外還有互聯(lián)網(wǎng)金融、安全風(fēng)控、與友商比價等業(yè)務(wù)，以及Logview、Mercury、Titan作為內(nèi)部服務(wù)的監(jiān)控系統(tǒng)、VDRC實時數(shù)據(jù)同步系統(tǒng)等。

實時平臺的職責(zé)主要包括實時計算平臺和實時基礎(chǔ)數(shù)據(jù)。實時計算平臺在Storm、Spark、Flink等計算框架的基礎(chǔ)上，為監(jiān)控、穩(wěn)定性提供了保障，為業(yè)務(wù)開發(fā)提供了數(shù)據(jù)的輸入與輸出。實時基礎(chǔ)數(shù)據(jù)包含對上游埋點的定義和規(guī)范化，對用戶行為數(shù)據(jù)、MySQL的Binlog日志等數(shù)據(jù)進行清洗、打?qū)挼忍幚?，為下游提供質(zhì)量保證的數(shù)據(jù)。

在架構(gòu)設(shè)計上，包括兩大數(shù)據(jù)源。一種是在App、微信、H5等應(yīng)用上的埋點數(shù)據(jù)，原始數(shù)據(jù)收集后發(fā)送到在kafka中;另一種是線上實時數(shù)據(jù)的MySQL Binlog日志。數(shù)據(jù)在計算框架里面做清洗關(guān)聯(lián)，把原始的數(shù)據(jù)通過實時ETL為下游的業(yè)務(wù)應(yīng)用(包括離線寬表等)提供更易于使用的數(shù)據(jù)。

Flink在唯品會的實踐

場景一：Dataeye實時看板

Dataeye實時看板是支持需要對所有的埋點數(shù)據(jù)、訂單數(shù)據(jù)等進行實時計算時，具有數(shù)據(jù)量大的特點，并且需要統(tǒng)計的維度有很多，例如全站、二級平臺、部類、檔期、人群、活動、時間維度等，提高了計算的復(fù)雜程度，統(tǒng)計的數(shù)據(jù)輸出指標(biāo)每秒鐘可以達到幾十萬。

以UV計算為例，首先對Kafka內(nèi)的埋點數(shù)據(jù)進行清洗，然后與Redis數(shù)據(jù)進行關(guān)聯(lián)，關(guān)聯(lián)好的數(shù)據(jù)寫入Kafka中;后續(xù)Flink計算任務(wù)消費Kafka的關(guān)聯(lián)數(shù)據(jù)。通常任務(wù)的計算結(jié)果的量也很大(由于計算維度和指標(biāo)特別多，可以達到上千萬)，數(shù)據(jù)輸出通過也是通過Kafka作為緩沖，最終使用同步任務(wù)同步到HBase中，作為實時數(shù)據(jù)展示。同步任務(wù)會對寫入HBase的數(shù)據(jù)限流和同類型的指標(biāo)合并，保護HBase。與此同時還有另一路計算方案作為容災(zāi)。 

在以Storm進行計算引擎中進行計算時，需要使用Redis作為中間狀態(tài)的存儲，而切換到Flink后，F(xiàn)link自身具備狀態(tài)存儲，節(jié)省了存儲空間;由于不需要訪問Redis，也提升了性能，整體資源消耗降低到了原來的1/3。

在將計算任務(wù)從Storm逐步遷移到Flink的過程中，對兩路方案先后進行遷移，同時將計算任務(wù)和同步任務(wù)分離，緩解了數(shù)據(jù)寫入HBase的壓力。

切換到Flink后也需要對一些問題進行追蹤和改進。對于FlinkKafkaConsumer，由于業(yè)務(wù)原因?qū)afka中的Aotu Commit進行修改，以及對offset的設(shè)定，需要自己實現(xiàn)支持kafka集群切換的功能。對不帶window的state數(shù)據(jù)需要手動清理。還有計算框架的通病——數(shù)據(jù)傾斜問題需要處理。同時對于同步任務(wù)追數(shù)問題，Storm可以從Redis中取值，F(xiàn)link只能等待。

場景二：Kafka數(shù)據(jù)落地HDFS

之前都是通過Spark Streaming的方式去實現(xiàn)，現(xiàn)在正在逐步切換到Flink上面，通過OrcBucketingTableSink將埋點數(shù)據(jù)落地到HDFS上的Hive表中。在Flink處理中單Task Write可達到3.5K/s左右，使用Flink后資源消耗降低了90%，同時將延遲30s降低到了3s以內(nèi)。目前還在做Flink對Spark Bucket Table的支持。

場景三：實時的ETL

對于ETL處理工作而言，存在的一個痛點就是字典表存儲在HDFS中，并且是不斷變化的，而實時的數(shù)據(jù)流需要與字典表進行join。字典表的變化是由離線批處理任務(wù)引起的，目前的做法是使用ContinuousFileMonitoringFunction和ContinuousFileReaderOperator定時監(jiān)聽HDFS數(shù)據(jù)變化，不斷地將新數(shù)據(jù)刷入，使用***的數(shù)據(jù)去做join實時數(shù)據(jù)。

我們計劃做更加通用的方式，去支持Hive表和Stream的join，實現(xiàn)Hive表數(shù)據(jù)變化之后，數(shù)據(jù)自動推送的效果。

Flink On K8S

在唯品會內(nèi)部有一些不同的計算框架，有實時計算的，有機器學(xué)習(xí)的，還有離線計算的，所以需要一個統(tǒng)一的底層框架來進行管理，因此將Flink遷移到了K8S上。

在K8S上使用了思科的網(wǎng)絡(luò)組件，每個docker容器都有獨立的ip，對外也是可見的。實時平臺的融合器整體架構(gòu)如下圖所示。

唯品會在K8S上的實現(xiàn)方案與Flink社區(qū)提供的方案差異還是很大的。唯品會使用K8S StatefulSet模式部署，內(nèi)部實現(xiàn)了cluster相關(guān)的一些接口。一個job對應(yīng)一個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腳本里面需要設(shè)置的環(huán)境變量設(shè)置說明：

| 環(huán)境變量名稱 | 參數(shù) | 示例內(nèi)容 | 說明 | |--- |---|---|---|---| | 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設(shè)置和hdfs checkpiont的根目錄 | | TASK_MANAGER_NUMBER_OF_TASK_SLOTS | containers.resources.cpu.limits | 2 | TM的slot數(shù)量，根據(jù)resources.cpu.limits來設(shè)置 | | 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大小，由于存在堆外內(nèi)存，需要小于container.resources.memory.limits;否則容易OOM kill | | TASK_MANAGER_HEAP_MB | env:TASK_MANAGER_HEAP_MB value: containers.resources.memory.limit -1024 |4096 | JM的Heap大小，由于存在堆外內(nèi)存，需要小于container.resources.memory.limits;否則容易OOM kill |

對應(yīng)Flink集群所依賴的HDFS等其他配置，則通過創(chuàng)建configmap來管理和維護。

kubectl create configmap hdfs-conf --from-file=hdfs-site.xml --from-file=core-site.xml 

后續(xù)計劃

當(dāng)前實時系統(tǒng)，機器學(xué)習(xí)平臺要處理的數(shù)據(jù)分布在各種數(shù)據(jù)存儲組件中，如Kafka、Redis、Tair和HDFS等，如何方便高效的訪問，處理，共享這些數(shù)據(jù)是一個很大的挑戰(zhàn)，對于當(dāng)前的數(shù)據(jù)訪問和解析常常需要耗費很多的精力，主要的痛點包括：

• 對于Kafka，Redis，Tair中的binary(PB/Avro等格式)數(shù)據(jù)，使用者無法快速直接的了解數(shù)據(jù)的schema與數(shù)據(jù)內(nèi)容，采集數(shù)據(jù)內(nèi)容及與寫入者的溝通成本很高。
• 由于缺少獨立的統(tǒng)一數(shù)據(jù)系統(tǒng)服務(wù)，對Kafka，Redis，Tair等中的binary數(shù)據(jù)訪問需要依賴寫入者提供的信息，如proto生成類，數(shù)據(jù)格式wiki定義等，維護成本高，容易出錯。
• 缺乏relational schema使得使用者無法直接基于更高效易用的SQL或LINQ層API開發(fā)業(yè)務(wù)。
• 無法通過一個獨立的服務(wù)方便的發(fā)布和共享數(shù)據(jù)。
• 實時數(shù)據(jù)無法直接提供給Batch SQL引擎使用。
• 此外，對于當(dāng)前大部分的數(shù)據(jù)源的訪問也缺少審計，權(quán)限管理，訪問監(jiān)控，跟蹤等特性。

UDM(統(tǒng)一數(shù)據(jù)管理系統(tǒng))包括Location Manager, Schema Metastore以及Client Proxy等模塊，主要的功能包括：

• 提供從名字到地址的映射服務(wù)，使用者通過抽象名字而不是具體地址訪問數(shù)據(jù)。
• 用戶可以方便的通過Web GUI界面方便的查看數(shù)據(jù)Schema，探查數(shù)據(jù)內(nèi)容。
• 提供支持審計，監(jiān)控，溯源等附加功能的Client API Proxy。
• 在Spark/Flink/Storm等框架中，以最適合使用的形式提供這些數(shù)據(jù)源的封裝。

UDM的整體架構(gòu)如下圖所示：

UDM的使用者包括實時，機器學(xué)習(xí)以及離線平臺中數(shù)據(jù)的生產(chǎn)者和使用者。在使用Sql API或Table API的時候，首先完成Schema的注冊，之后使用Sql進行開發(fā)，降低了開發(fā)代碼量。

在Flink中，使用UDMExternalCatalog來打通Flink計算框架和UDM之間的橋梁，通過實現(xiàn)ExternalCatalog的各個接口，以及實現(xiàn)各自數(shù)據(jù)源的TableSourceFactory，完成Schema和接入管控等各項功能。

關(guān)于作者：王新春目前在唯品會負(fù)責(zé)實時平臺相關(guān)內(nèi)容，主要包括實時計算框架和提供實時基礎(chǔ)數(shù)據(jù)，以及機器學(xué)習(xí)平臺的工作。之前在美團點評，也是負(fù)責(zé)大數(shù)據(jù)平臺工作。他已經(jīng)在大數(shù)據(jù)實時處理方向積累了豐富的工作經(jīng)驗。