摘要：Apache Flink 是一個分布式大數(shù)據(jù)處理引擎，可對有限數(shù)據(jù)流和無限數(shù)據(jù)流進(jìn)行有狀態(tài)計(jì)算。本文由貝殼找房的資深工程師劉力云將帶來Apache Flink技術(shù)在貝殼找房業(yè)務(wù)中的應(yīng)用，通過企業(yè)開發(fā)的實(shí)時計(jì)算平臺案例的分享幫助用戶了解Apache Flink的技術(shù)特性與應(yīng)用場景。

業(yè)務(wù)規(guī)模及演進(jìn)

下圖為貝殼找房的業(yè)務(wù)場景示意圖。最上層為貝殼找房公司最為主體的四大業(yè)務(wù)：二手房交易、新房交易、租賃業(yè)務(wù)及裝修業(yè)務(wù)。四大業(yè)務(wù)運(yùn)營將產(chǎn)生圖示中間部分的四大數(shù)據(jù)即樓盤字典、交易數(shù)據(jù)、用戶行為日志與后端服務(wù)日志。圖示最下部分代表公司實(shí)時數(shù)據(jù)采集、實(shí)時數(shù)據(jù)計(jì)算的業(yè)務(wù)模塊，本文中的案例將重點(diǎn)介紹數(shù)據(jù)實(shí)時計(jì)算部分的設(shè)計(jì)、實(shí)現(xiàn)及應(yīng)用內(nèi)容。

發(fā)展歷程

在2018年初，隨著公司埋點(diǎn)治理規(guī)范的推進(jìn)，我們建設(shè)了DP實(shí)時數(shù)據(jù)總線，統(tǒng)一承接各種埋點(diǎn)數(shù)據(jù)流的標(biāo)準(zhǔn)化處理，并對外提供清洗后的實(shí)時數(shù)據(jù)。隨著維護(hù)的實(shí)時任務(wù)增加，面臨著實(shí)時數(shù)據(jù)流穩(wěn)定性以及任務(wù)管理方面的挑戰(zhàn)，于是貝殼大數(shù)據(jù)部著手研發(fā)了Hermes實(shí)時計(jì)算平臺，提供統(tǒng)一的實(shí)時任務(wù)管理平臺。

在2018年10月，我們推出了SQL V1編輯器來方便用戶開發(fā)實(shí)時計(jì)算任務(wù)。SQL V1基于Spark Structured Streaming技術(shù)，用戶可以使用SQL完成需求的開發(fā)，同時以界面拖拽的形式呈現(xiàn)給用戶，使用戶的操作更加便捷。在2019年5月，經(jīng)過調(diào)研對比，我們引入了Flink技術(shù)棧，研發(fā)的SQL V2編輯器正式上線，SQL V2全面支持Flink SQL的各種語法并設(shè)計(jì)了大量的自定義函數(shù)，兼容hive UDF以及用戶常用函數(shù)。目前我們已經(jīng)在公司內(nèi)進(jìn)行實(shí)時數(shù)倉業(yè)務(wù)場景的探索應(yīng)用。

應(yīng)用規(guī)模

下圖所示為目前實(shí)時計(jì)算在貝殼找房企業(yè)中的應(yīng)用規(guī)模。目前平臺支持30余個業(yè)務(wù)項(xiàng)目，流計(jì)算任務(wù)數(shù)達(dá)到400個，隨著數(shù)倉的不斷擴(kuò)充，實(shí)時流計(jì)算的任務(wù)數(shù)將不斷上升。每日處理的消息條數(shù)達(dá)到了800億級別，效率十分可觀。

支持的項(xiàng)目

從下圖所示實(shí)時計(jì)算在企業(yè)中的支持項(xiàng)目可以看出，目前實(shí)時計(jì)算平臺支持從風(fēng)控、租賃到策略搜索再到新房交易等一系列業(yè)務(wù)項(xiàng)目，從各個維度支持起了企業(yè)運(yùn)營產(chǎn)生出數(shù)據(jù)實(shí)時計(jì)算業(yè)務(wù)需求。

任務(wù)增長趨勢

最開始平臺上線時支持的任務(wù)增長較為緩慢，在2019年6月初，平臺升級到Flink并全面支持SQL開發(fā)后，任務(wù)數(shù)量開始大規(guī)模的增長，在2019年11月份實(shí)時數(shù)倉建成后，平臺所支持的任務(wù)數(shù)量有了十分明顯的增長趨勢。

平臺數(shù)據(jù)統(tǒng)計(jì)

下圖所示為平臺每日數(shù)據(jù)統(tǒng)計(jì)。目前平臺每日可以處理1000億條數(shù)據(jù)，一般數(shù)據(jù)任務(wù)的處理延遲在40毫秒左右。

Hermes實(shí)時計(jì)算平臺介紹

平臺概覽

Hermes平臺目前支持著公司實(shí)時任務(wù)的開發(fā)、編輯、部署、啟停等管理功能及豐富的監(jiān)控報(bào)警等服務(wù)。平臺支持Java、Scala、Python等多種語言開發(fā)的實(shí)時任務(wù)，支持自定義任務(wù)、模板任務(wù)及場景任務(wù)三大任務(wù)類型，同時做到了各個項(xiàng)目的資源隔離，每個項(xiàng)目均有項(xiàng)目的專有隊(duì)列，防止與其他項(xiàng)目在資源上發(fā)生競爭。平臺同時為資源需求較小的項(xiàng)目提供了公共隊(duì)列，通過公共隊(duì)列對該種項(xiàng)目進(jìn)行支持的方式，更為方便的實(shí)現(xiàn)任務(wù)的開發(fā)。

Hermes架構(gòu)

下圖所示為Hermes平臺的整體架構(gòu)，架構(gòu)分為4個層次，圖中最下層深藍(lán)條目代表架構(gòu)中的計(jì)算引擎，目前計(jì)算引擎支持Flink與Spark Streaming技術(shù)，并通過消息隊(duì)列、離線存儲等技術(shù)輔助完成數(shù)據(jù)實(shí)時的存儲。在引擎層方面，架構(gòu)采用StreamSQL、DataStream、StreamCEP等技術(shù)搭建，其中StreamCEP技術(shù)很好的支持了經(jīng)紀(jì)人平臺業(yè)務(wù)實(shí)時監(jiān)控報(bào)警的需求。功能組件層方面包括了任務(wù)實(shí)例的管理、項(xiàng)目管理及數(shù)據(jù)源管理等。平臺目前可以在同一任務(wù)中的不同任務(wù)快照間進(jìn)行相互切換，當(dāng)發(fā)現(xiàn)上線任務(wù)有問題時，可以回退到之前的快照。

SQL V1編輯器

下圖所示為SQL V1編輯器示意圖。該編輯器對于大部分?jǐn)?shù)據(jù)清洗及數(shù)據(jù)處理的業(yè)務(wù)場景可以實(shí)現(xiàn)簡潔高效的編輯處理。用戶在編輯器左側(cè)可以定義編輯數(shù)據(jù)源、操作符及目標(biāo)源等數(shù)據(jù)信息。中央面板上呈現(xiàn)的數(shù)據(jù)為SQL V1支持編輯的操作類型，選中面板中央的過濾器，即可在編輯器右側(cè)添加相關(guān)的過濾條件，實(shí)現(xiàn)數(shù)據(jù)的相關(guān)過濾。在目標(biāo)源層面，編輯器目前支持Kafka、Druid等多種目標(biāo)源，大大提升了編輯器的兼容性。

SQL V2編輯器

下圖所示為SQL V2編輯器示意圖。目前SQL V2是基于Flink SQL技術(shù)較為完善的編輯器，左側(cè)為用戶進(jìn)行代碼編輯的部分，用戶在此處可以編輯大量SQL語句以此助力不同業(yè)務(wù)場景。左下欄目中的數(shù)據(jù)為用戶選中數(shù)據(jù)源自動生成的DDL，通過DDL編輯器將操作數(shù)據(jù)的樣式更清晰的展示給用戶。SQL V2支持了三大類型的數(shù)據(jù)表，分別是source表、sink表及維表，以此方便用戶的開發(fā)。編輯器右下角可以呈現(xiàn)SQL語法的檢測情況，以此提示用戶在編輯時出現(xiàn)的語法錯誤。

SQL V2架構(gòu)

SQL V2工具整體架構(gòu)如下圖所示。前端SQL編輯器模塊包括語法語義的檢查、執(zhí)行計(jì)劃的查看、自動DDL的生成及任務(wù)調(diào)試的功能。用戶通過任務(wù)調(diào)試功能可以查看任務(wù)執(zhí)行結(jié)果。后臺將引擎提交到Y(jié)arn集群上執(zhí)行，引擎通過任務(wù)id回調(diào)后臺接口獲取需要執(zhí)行的SQL，對SQL做語法校驗(yàn)和語法解析，若出現(xiàn)維表關(guān)聯(lián)則會額外對SQL做一層轉(zhuǎn)換。

SQL引擎整體架構(gòu)

下圖所示為SQL引擎的整體架構(gòu)。整體架構(gòu)分為三個層次，最底層為Flink Table API。在Flink層之上企業(yè)設(shè)計(jì)了代碼的封裝，以factory的形式方便最上層的方法調(diào)用。最上層的core層負(fù)責(zé)整個系統(tǒng)的SQL解析。

維表關(guān)聯(lián)

在SQL解析過程中，最為復(fù)雜的是維表的表格關(guān)聯(lián)，下圖為維表關(guān)聯(lián)系統(tǒng)架構(gòu)圖。數(shù)據(jù)從數(shù)據(jù)源導(dǎo)入后，系統(tǒng)使用Async I/O技術(shù)訪問后端，系統(tǒng)后端使用Data Accessor接口訪問后端的存儲。系統(tǒng)后端存儲支持HBase與Redis存儲技術(shù)，同時后端會將數(shù)據(jù)緩存于LRU Cache模塊中。維表關(guān)聯(lián)后的數(shù)據(jù)支持多種大數(shù)據(jù)工具的存儲，從而大大增加了系統(tǒng)的兼容性。

豐富的內(nèi)置函數(shù)

系統(tǒng)同時為用戶提供了豐富的內(nèi)置函數(shù)，包括時間函數(shù)、集合函數(shù)、Json處理函數(shù)及字符串函數(shù)。豐富的內(nèi)置函數(shù)可以方便用戶的開發(fā)，省去用戶自己去開發(fā)的時間。

實(shí)時數(shù)倉整體架構(gòu)

下圖所示為實(shí)時數(shù)倉的整體架構(gòu)，同時也是SQL V2系統(tǒng)落地的應(yīng)用場景。各個層級間產(chǎn)生的數(shù)據(jù)被儲存在了Kafka Topic中，同時數(shù)據(jù)也將被同步到hive中備份。業(yè)務(wù)方可以查詢實(shí)時備份數(shù)據(jù)進(jìn)行數(shù)據(jù)驗(yàn)證及分析等操作。目前數(shù)倉的實(shí)時計(jì)算部分可以計(jì)算當(dāng)天或過往幾天的數(shù)據(jù)，實(shí)時計(jì)算平臺正在與其他組件合作，開發(fā)實(shí)時與離線聯(lián)合的分析查詢，以此擴(kuò)展實(shí)時數(shù)倉的使用范圍。

實(shí)時數(shù)倉數(shù)據(jù)統(tǒng)計(jì)

下圖所示為企業(yè)實(shí)時數(shù)倉的數(shù)據(jù)統(tǒng)計(jì)。從2019年8月，SQL V2正式上線運(yùn)營，至2019年10月平臺開始支持實(shí)時數(shù)倉開發(fā)，系統(tǒng)的數(shù)據(jù)量開始加速增長。目前，實(shí)時數(shù)倉已經(jīng)有100余個任務(wù)，數(shù)據(jù)吞吐量也達(dá)到了21億條/天的數(shù)據(jù)級別，數(shù)據(jù)規(guī)模較為可觀。

實(shí)時數(shù)倉案例

下圖列舉出實(shí)時數(shù)倉平臺已經(jīng)實(shí)現(xiàn)提供數(shù)據(jù)支持的應(yīng)用案例。

1. 交易平臺

交易平臺實(shí)時大屏實(shí)時展示大區(qū)內(nèi)的交易狀況。在交易平臺的建設(shè)中，開發(fā)團(tuán)隊(duì)通過數(shù)據(jù)回環(huán)將還未關(guān)聯(lián)的數(shù)據(jù)返回儲存模塊進(jìn)行重新關(guān)聯(lián)，并通過檢驗(yàn)該數(shù)據(jù)的生命周期判斷是否關(guān)聯(lián)成功，團(tuán)隊(duì)通過此種方式使得數(shù)據(jù)維表與事實(shí)表數(shù)據(jù)最終一致。

2. 經(jīng)紀(jì)人行程量

經(jīng)紀(jì)人行程量可以動態(tài)的展示當(dāng)前經(jīng)紀(jì)人對客戶的維護(hù)情況，使企業(yè)可以掌握經(jīng)紀(jì)人實(shí)時的工作狀態(tài)。

3. 實(shí)時用戶畫像

實(shí)時用戶畫像可以實(shí)時地向企業(yè)呈現(xiàn)來自各個系統(tǒng)用戶的數(shù)據(jù)信息，通過組合各個平臺上用戶的行為信息，提供全面、精準(zhǔn)的用戶畫像。企業(yè)的算法策略部門將根據(jù)用戶的實(shí)時畫像進(jìn)行相關(guān)信息、內(nèi)容的推薦。

監(jiān)控報(bào)警

下圖為平臺的監(jiān)控報(bào)警頁面截圖。監(jiān)控系統(tǒng)會實(shí)時監(jiān)控平臺任務(wù)的處理延時、source寫入量及sink寫出量三大指標(biāo)。系統(tǒng)中同時可以設(shè)置平臺數(shù)據(jù)的無心跳時間，當(dāng)超出設(shè)置時限后，系統(tǒng)將會進(jìn)行報(bào)警。

監(jiān)控報(bào)警架構(gòu)

下圖為監(jiān)控報(bào)警架構(gòu)圖。監(jiān)控系統(tǒng)通過自定義的Listener對Spark進(jìn)行監(jiān)控，Listener引入SDK收集Spark任務(wù)的信息及運(yùn)行中的日志數(shù)據(jù)。用戶在此處需要進(jìn)行手動SDK的導(dǎo)入。在Flink應(yīng)用模塊中，系統(tǒng)設(shè)計(jì)支持了自定義Report數(shù)據(jù)的獲取，并通過自動加載的方式直接載入Flink中進(jìn)行數(shù)據(jù)的分析與計(jì)算，同時通過任務(wù)啟動是注入java探針的方式獲取任務(wù)的相關(guān)信息。所有的監(jiān)控信息將被統(tǒng)一送到Kafka Topic中，經(jīng)Hermes平臺分析處理，觸發(fā)相應(yīng)的延時報(bào)警及心跳報(bào)警。

未來發(fā)展與規(guī)劃

整體架構(gòu)

實(shí)時計(jì)算平臺的整體架構(gòu)如下圖所示。在架構(gòu)中間部分，平臺包含了實(shí)時事件中心、事件處理平臺等系統(tǒng)來更好的處理未來企業(yè)中的業(yè)務(wù)場景需求，以通用服務(wù)平臺的方式為更多的業(yè)務(wù)方提供統(tǒng)一的業(yè)務(wù)支撐。在引擎方面，未來會深入研究Flink的狀態(tài)管理、端到端的精確一次等技術(shù)，提高數(shù)據(jù)處理的準(zhǔn)確性和一致性。