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

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

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

發(fā)展歷程

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

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

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

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

支持的項(xiàng)目

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

任務(wù)增長(zhǎng)趨勢(shì)

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

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

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

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

平臺(tái)概覽

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

Hermes架構(gòu)

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

SQL V1編輯器

下圖所示為SQL V1編輯器示意圖。該編輯器對(duì)于大部分?jǐn)?shù)據(jù)清洗及數(shù)據(jù)處理的業(yè)務(wù)場(chǎng)景可以實(shí)現(xiàn)簡(jiǎ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ǔ)句以此助力不同業(yè)務(wù)場(chǎng)景。左下欄目中的數(shù)據(jù)為用戶選中數(shù)據(jù)源自動(dòng)生成的DDL，通過DDL編輯器將操作數(shù)據(jù)的樣式更清晰的展示給用戶。SQL V2支持了三大類型的數(shù)據(jù)表，分別是source表、sink表及維表，以此方便用戶的開發(fā)。編輯器右下角可以呈現(xiàn)SQL語(yǔ)法的檢測(cè)情況，以此提示用戶在編輯時(shí)出現(xiàn)的語(yǔ)法錯(cuò)誤。

SQL V2架構(gòu)

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

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

下圖所示為SQL引擎的整體架構(gòu)。整體架構(gòu)分為三個(gè)層次，最底層為Flink Table API。在Flink層之上企業(yè)設(shè)計(jì)了代碼的封裝，以factory的形式方便最上層的方法調(diào)用。最上層的core層負(fù)責(zé)整個(gè)系統(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接口訪問后端的存儲(chǔ)。系統(tǒng)后端存儲(chǔ)支持HBase與Redis存儲(chǔ)技術(shù)，同時(shí)后端會(huì)將數(shù)據(jù)緩存于LRU Cache模塊中。維表關(guān)聯(lián)后的數(shù)據(jù)支持多種大數(shù)據(jù)工具的存儲(chǔ)，從而大大增加了系統(tǒng)的兼容性。

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

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

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

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

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

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

實(shí)時(shí)數(shù)倉(cāng)案例

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

1. 交易平臺(tái)

交易平臺(tái)實(shí)時(shí)大屏實(shí)時(shí)展示大區(qū)內(nèi)的交易狀況。在交易平臺(tái)的建設(shè)中，開發(fā)團(tuán)隊(duì)通過數(shù)據(jù)回環(huán)將還未關(guān)聯(lián)的數(shù)據(jù)返回儲(chǔ)存模塊進(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ì)人行程量可以動(dòng)態(tài)的展示當(dāng)前經(jīng)紀(jì)人對(duì)客戶的維護(hù)情況，使企業(yè)可以掌握經(jīng)紀(jì)人實(shí)時(shí)的工作狀態(tài)。

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

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

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

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

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

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

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

整體架構(gòu)

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