背景

58同城作為覆蓋生活全領域的服務平臺，業務覆蓋招聘、房產、汽車、金融、二手及本地服務等各個方面。 豐富的業務線和龐大的用戶數每天產生海量用戶數據需要實時化的計算分析，實時計算平臺定位于為集團海量數據提供高效、穩定、分布式實時計算的基礎服務。 本文主要介紹58同城基于Flink打造的一站式實時計算平臺Wstream。

實時計算場景

和很多互聯網公司一樣，實時計算在58擁有豐富的場景需求，主要包括以下幾類：

1.實時數據ETL

實時消費Kafka數據進行清洗、轉換、結構化處理用于下游計算處理。

2.實時數倉

實時化數據計算，倉庫模型加工和存儲。 實時分析業務及用戶各類指標，讓運營更加實時化。

3.實時監控

對系統和用戶行為進行實時檢測和分析，如業務指標實時監控，運維線上穩定性監控，金融 風控等。

4.實時分析

特征平臺，用戶畫像，實時個性化推薦等。

平臺演進

在實時計算平臺建設過程中，主要是跟進開源社區發展以及實際業務需求，計算框架經歷了Storm到 Spark Streaming到 Flink的發展，同時建設一站式實時計算平臺，旨在提升用戶實時計算需求開發上線管理監控效率，優化平臺管理。

實時計算引擎前期基于Storm和Spark Streaming構建,很多情況下并不能很好的滿足業務需求，如商業部門基于Spark Streaming構建的特征平臺希望將計算延遲由分鐘級降低到秒級，提升用戶體驗，運維監控平臺基于Storm分析公司全量nginx日志對線上業務進行監控，需要秒級甚至毫秒級別的延遲，Storm的吞吐能力成為瓶頸。 同時隨著實時需求不斷增加，場景更加豐富，在追求任務高吞吐低延遲的基礎上，對計算過程中間狀態管理，靈活窗口支持，以及exactly once語義保障的訴求越來越多。 Apache Flink開源之后，支持高吞吐低延遲的架構設計以及高可用的穩定性，同時擁有實時計算場景一系列特性以及支持實時Sql模型，使我們決定采用 Flink作為新一代實時計算平臺的計算引擎。

平臺規模

實時計算平臺當前主要基于Storm/Spark Streaming/Flink,集群共計500多臺機器，每天處理數據量6000億+，其中Flink經過近一年的建設，任務占比已經達到50% 。

Flink穩定性

Flink作為實時計算集群，可用性要求遠高于離線計算集群。 為保障集群可用性，平臺主要采用任務隔離以及高可用集群架構保障穩定性。

任務隔離  

在應用層面主要基于業務線以及場景進行機器隔離，隊列資源分配管理，避免集群抖動造成全局影響。

集群架構

Flink集群采用了ON YARN模式獨立部署，為減少集群維護工作量，底層HDFS利用公司統一HDFS Federation架構下建立獨立的namespace，減少Flink任務在checkpoint采用hdfs/rocksdb作為狀態存儲后端場景下由于hdfs抖動出現頻繁異常失敗。 在資源隔離層面，引入Node Label機制實現重要任務運行在獨立機器，不同計算性質任務運行在合適的機器下，最大化機器資源的利用率。 同時在YARN資源隔離基礎上增加Cgroup進行物理cpu隔離，減少任務間搶占影響，保障任務運行穩定性。

平臺化管理

Wstream是一套基于Apache Flink構建的一站式、高性能實時大數據處理平臺。 提供SQL化流式數據分析能力，大幅降低數據實時分析門檻，支持通過DDL實現source/sink以及維表，支持UDF/UDAF/UDTF，為用戶提供更強大的數據實時處理能力。 支持多樣式應用構建方式FlinkJar/Stream SQL/Flink-Storm，以滿足不同用戶的開發需求，同時通過調試，監控，診斷，探查結果等輔助手段完善任務生命周期管理。

流式sql能力建設

Stream SQL是平臺為了打造sql化實時計算能力，減小實時計算開發門檻，基于開源的 Flink，對底層sql模塊進行擴展實現以 下功能

 1.支持自定義DDL語法（包括源表,輸出表,維表）

 2.支持自定義UDF/UDTF/UDAF語法

 3.實現了流與維表的join,雙流join

在支持大數據開源組件的同時，也打通了公司主流的實時存儲平臺。 同時為用戶提供基于Sql client的cli方式以及在Wstream集成了對實時sql能力的支持，為用戶提供在線開發調試sql任務的編輯器，同時支持代碼高亮，智能提示，語法校驗及運行時校驗，盡可能避免用戶提交到集群的任務出現異常。 另外也為用戶提供了向導化配置方式，解決用戶定義table需要了解復雜的參數設置，用戶只需關心業務邏輯處理，像開發離線Hive一樣使用sql開發實時任務。

Storm任務遷移Flink

在完善Flink平臺建設的同時，我們也啟動Storm任務遷移Flink計劃，旨在提升實時計算平臺整體效率，減少機器成本和運維成本。 Flink-Storm作為官方提供Flink兼容Storm程序為我們實現無縫遷移提供了可行性，但是作為beta版本，在實際使用過程中存在很多無法滿足現實場景的情況，因此我們進行了大量改進，主要包括實現Storm任務on yarn ，遷移之后任務at least once語義保障，兼容Storm的 tick tuple機制等等。

通過對Fink-Storm的優化，在無需用戶修改代碼的基礎上，我們已經順利完成多個Storm版本集群任務遷移和集群下線，在保障實時性及吞吐量的基礎上可以節約計算資源40%以上，同時借助yarn統一管理實時計算平臺無需維護多套Storm集群，整體提升了平臺資源利用率，減輕平臺運維工作量。

任務診斷 

指標監控

Flink webUI 提供了大量的運行時信息供用戶了解任務當前運行狀況，但是存在無法獲取歷史metrics的問題導致用戶無法了解任務歷史運行狀態，因此我們采用了Flink原生支持的Prometheus進行實時指標采集和存儲，Prometheus是一個開源的監控和報警系統，通過pushgateway的方式實時上報metrics，Prometheus集群采用Fedration部署模式，meta節點定時抓取所有子節點指標進行匯總，方便統一數據源提供給Grafana進行可視化以及告警配置。

任務延遲

吞吐能力和延遲作為衡量實時任務性能最重要的指標，我們經常需要通過這兩個指標來調整任務并發度和資源配置。 Flink Metrics提供latencyTrackingInterval參數啟用任務延遲跟蹤，打開會顯著影響集群和任務性能，官方高度建議只在debug下使用。 在實踐場景下，Flink任務數據源基本都是Kafka，因此我們采用topic消費堆積作為衡量任務延遲的指標，監控模塊實時通過Flink rest獲取任務正在消費topic的offset，同時通過Kafka JMX獲取對應topic的logsize，采用logsize– offset作為topic的堆積。

日志檢索

Flink 作為分布式計算引擎，所有任務會由YARN統一調度到任意的計算節點，因此任務的運行日志會分布在不同的機器，用戶定位日志困難，我們通過調整log4j日志框架默認機制，按天切分任務日志，定期清理過期日志，避免異常任務頻繁寫滿磁盤導致計算節點不可用的情況，同時在所有計算節點部署agent 實時采集日志，匯聚寫入Kafka，通過日志分發平臺實時將數據分發到ES，方便用戶進行日志檢索和定位問題。

Flink優化

在實際使用過程中， 我們也針對業務場景進行了一些優化和擴展，主要包括：

1.Storm任務需要Storm引擎提供ack機制保障消息傳遞at least once語義，遷移到Flink無法使用ack機制，我們通過定制KafakSpout實現checkpoint相關接口，通過Flink checkpoint機制實現消息傳遞不丟失。 另外Flink-Storm默認只能支持standalone的提交方式，我們通過實現yarn client相關接口增加了storm on yarn的支持。

2.Flink 1.6推薦的是一個TaskManager對應一個slot的使用方式，在申請資源的時候根據最大并發度申請對應數量的TaskManger，這樣導致的問題就是在任務設置task slots之后需要申請的資源大于實際資源。 我們通過在ResoureManager請求資源管理器SlotManager的時候增加TaskManagerSlot相關信息 ，用于維護申請到的待分配TaskManager和slot，之后對于SlotRequests請求不是直接申請TaskManager，而是先從SlotManager申請是否有足夠slot，沒有才會啟動新的TaskManger,這樣就實現了申請資源等于實際消耗資源，避免任務在資源足夠的情況下無法啟動。

3.Kafak Connector改造，增加自動換行支持，另外針對08source無法設置client.id，通過將client.id生成機制優化成更有標識意義的id，便于Kafka層面管控

4.Flink提交任務無法支持第三方依賴jar包和配置文件供TaskManager使用，我們通過修改flink啟動腳本，增加相關參數支持外部傳輸文件，之后在任務啟動過程中通過將對應的jar包和文件加入classpath，借助yarn的文件管理機制實現類似spark對應的使用方式，方便用戶使用

5.業務場景存在大量實時寫入hdfs需求，Flink 自帶BucketingSink默認只支持string和avro格式，我們在此基礎上同時支持了LZO及Parquet格式寫入，極大提升數據寫入性能。

后續規劃

實時計算平臺當前正在進行Storm任務遷移Flink集群，目前已經基本完成，大幅提升了平臺資源利用率和計算效率。 后續將繼續調研完善Flink相關能力，推動Flink在更多的實時場景下的應用，包括實時規則引擎，實時機器學習等。