Flink 實踐 | 貝殼基于 Flink 的實時計算演進(jìn)之路

摘要：貝殼找房大數(shù)據(jù)平臺實時計算負(fù)責(zé)人劉力云帶來的分享內(nèi)容是貝殼找房的實時計算演進(jìn)之路，內(nèi)容如下：

1. 發(fā)展歷程
   

2. 平臺建設(shè)
   

3. 實時數(shù)倉及其應(yīng)用場景
   

4. 事件驅(qū)動場景
   

5. 未來規(guī)劃
   

Tips：點擊文末「閱讀原文」即可回顧作者原版分享視頻～

 

GitHub 地址 

https://github.com/apache/flink

歡迎大家給 Flink 點贊送 star~

一、發(fā)展歷程

首先是平臺的發(fā)展歷程。最早是因為業(yè)務(wù)方在實時計算方面有比較多的業(yè)務(wù)場景，包括業(yè)務(wù)方自研的實時任務(wù)，需要自行開發(fā)、部署及維護(hù)，我們的大數(shù)據(jù)部門也會承接客戶大數(shù)據(jù)的實時開發(fā)需求。

這些看起來都是一些煙囪式的開發(fā)架構(gòu)（即每個業(yè)務(wù)線之間由不同的開發(fā)團(tuán)隊獨立建設(shè)，技術(shù)棧不同，互不聯(lián)系），缺乏統(tǒng)一的任務(wù)管控，也很難保留開發(fā)過程中積累的技術(shù)沉淀。因此，我們在 18 年時上線了基于 Spark Streaming 的實時計算平臺，統(tǒng)一部署管理實時計算任務(wù)。之后我們又在此基礎(chǔ)上提供了任務(wù)開發(fā)功能 - 標(biāo)準(zhǔn)化的 SQL 語言（SQL 1.0），以提高數(shù)據(jù)開發(fā)效率。

隨著我們承接的任務(wù)越來越多，我們也發(fā)現(xiàn)了 Spark Streaming 的一些使用問題，主要是其 Checkpoint 是同步的，有時會造成比較大的延遲。此外，Kafka 消費的 Offset 數(shù)據(jù)存在 Checkpoint，很難做到任務(wù)細(xì)粒度的監(jiān)控，比如消費狀態(tài)的獲取，于是我們開始轉(zhuǎn)向 Flink。

19 年，我們的平臺開始支持 Flink 任務(wù)，并且很快提供了基于 Flink 1.8 的 SQL 2.0 功能，包括 DDL 定義和維表關(guān)聯(lián)。接下來，在 SQL 2.0  的基礎(chǔ)上，我們開始了實時數(shù)倉的建設(shè)。

今年初，在收集了業(yè)務(wù)方的需求場景后，我們認(rèn)為在實時事件處理方面需求明確，而且目前的實現(xiàn)也存在較多的弊端，因此我們開始著手事件處理平臺的開發(fā)。今年發(fā)布的 Flink 1.11 在 SQL 方面有很大的提升，我們在其基礎(chǔ)上正在開發(fā)一套統(tǒng)一的 SQL（3.0）。

目前平臺支持的部門涵蓋了貝殼絕大部分的業(yè)務(wù)方，支持各種場景，包括人店相關(guān)的房源、客源、經(jīng)紀(jì)人、風(fēng)控以及運(yùn)營等。

目前平臺支持的項目有 30 多個。在 SQL2.0 后，平臺上的任務(wù)數(shù)有明顯增長，達(dá)到 800 多個。由于貝殼所有的流量數(shù)據(jù)、用戶行為分析、以及數(shù)倉的建設(shè)都是通過平臺來構(gòu)建的，所以數(shù)據(jù)量很大，每天處理的消息達(dá) 2500 億條，單任務(wù)的消息吞吐量峰值達(dá) 3 百萬。

這是我們平臺任務(wù)的增長情況，可以明顯看到 19 年 10 月 SQL 2.0 上線且支持實時數(shù)倉開發(fā)后，任務(wù)增長勢頭顯著。

二、平臺建設(shè)

平臺的功能概覽包括四個方面：

• 支持任務(wù)托管的基本能力，包括任務(wù)的編輯發(fā)布、版本管理、監(jiān)控報警等；

  
  

• 支持多種語言的實時任務(wù)，包括對貝殼算法相關(guān)的 Python 實時任務(wù)的良好支持；

  
  

• 根據(jù)業(yè)務(wù)場景不同，支持多種業(yè)務(wù)類型，如自定義任務(wù)、模板任務(wù)及場景任務(wù)（SQL 任務(wù)），內(nèi)部通用配置化任務(wù)，如分流合并操作。目前 SQL 任務(wù)在平臺占比較高，我們的目標(biāo)是 80%；

  
  

• 支持公共隊列（針對較數(shù)據(jù)量小的需求），對于數(shù)據(jù)量大的需求，要有穩(wěn)定的資源保證，我們可以提供專有隊列，運(yùn)行更為可靠。
  

平臺的整體架構(gòu)與其它公司的差不多。底層是計算和存儲層，計算支持 Flink 和  Spark，主要包括消息隊列和各種 OLAP 存儲，同時也支持 MySQL，Hive 也可以做到實時落地，維表支持 Redis，HBase 存儲。ClickHouse 是目前主要的實時  OLAP 存儲，由于 Doris 支持 update，同時對關(guān)聯(lián)查詢的支持也比較好，我們也在嘗試 Doris 存儲。

引擎層主要封裝的是 SQL 引擎、DataStream 的通用性操作。在事件處理方面，對 Flink 的 CEP，包括對其它普通規(guī)則也做了較好的封裝。

開發(fā)管理層提供了各種任務(wù)的開發(fā)、監(jiān)控和資源管理。

平臺之上，也是提供了對 ETL、BI、推薦、監(jiān)控、風(fēng)控等各種業(yè)務(wù)場景的支持。

這是平臺任務(wù)生命周期的管理。可以看到，在啟動后會新建實例，從集群拿到運(yùn)行狀態(tài)后會判斷是否正常運(yùn)行。“是”則轉(zhuǎn)成運(yùn)行中狀態(tài)。在運(yùn)行過程中會對任務(wù)做延遲和心跳的監(jiān)控；如果說任務(wù)發(fā)生了異常，并且在配置中設(shè)置了延遲或心跳時長的閾值，則會嘗試進(jìn)行重啟。用戶可以在啟動任務(wù)時設(shè)置重啟次數(shù)，當(dāng)超過該值時，則認(rèn)為重啟失敗，將發(fā)送告警給任務(wù)負(fù)責(zé)人。

這是平臺監(jiān)控報警的架構(gòu)。我們在 Spark 引入了 sdk 依賴，在用戶開發(fā)任務(wù)時用代碼顯示添加就可以監(jiān)聽系統(tǒng)關(guān)心的指標(biāo)。Flink 任務(wù)支持自定義 Reporter 的 metrics 的獲取。我們還支持 java agent 的依賴注入，通過依賴注入我們可以獲取實時任務(wù)的制定信息。在 Hermes 平臺，我們可以拿到這些監(jiān)控信息，來支持延時報警、心跳報警、及數(shù)據(jù)血緣基礎(chǔ)上的流量分析，后續(xù)的有狀態(tài)任務(wù)恢復(fù)也依賴這些監(jiān)控指標(biāo)。監(jiān)控日志落入存儲（InfluxDB）之后可以進(jìn)行可視化處理，方便的查看歷史運(yùn)行狀態(tài)。

這是平臺監(jiān)控查看頁面，分別顯示了數(shù)據(jù)讀入、寫出、及延時的情況。

 

三、實時數(shù)倉

我們的實時數(shù)倉目前具備以下幾方面能力：首先是完善的元數(shù)據(jù)管理，包括連接管理和表管理；數(shù)倉開發(fā)人員共同構(gòu)建了數(shù)據(jù)分層架構(gòu)，包括 4 個分層：

• 在實時側(cè)，分層越少越好，否則中間環(huán)節(jié)越多，出問題的概率越大；

  
  

• 在 SQL 層面，支持標(biāo)準(zhǔn)的SQL語法，維表關(guān)聯(lián)，提供圖形化的SQL開發(fā)環(huán)境。另外還支持豐富的內(nèi)置函數(shù)，并逐步完善支持用戶自定義函數(shù)（UDF）的開發(fā)；

  
  

• 數(shù)據(jù)血緣方面，平臺支持圖形化展示和完善的鏈路分析，而且能實時看到數(shù)據(jù)流的運(yùn)行情況并對異常進(jìn)行標(biāo)示；

  
  

• 最后是多源支持，對公司內(nèi)部用到的各種存儲做到了較好的支持。

這是簡易的實時數(shù)倉架構(gòu)圖，總體來說是屬于 Lambda 架構(gòu)，包括實時流和離線流，以及離線流對實時流數(shù)據(jù)覆蓋的修復(fù)。從用戶行為日志、后端服務(wù)器日志及業(yè)務(wù)數(shù)據(jù)庫采集來的消息流，匯入并通過 ODS（Opertional Data Source）層再到 DW（Data Warehouse）層，我們支持 ODS 和 DW 層對維度進(jìn)行擴(kuò)充，關(guān)聯(lián)維表。

目前 DWD（Data Warehouse Detail）層的數(shù)據(jù)直接送入 ClickHouse，ClickHouse 現(xiàn)在是我們 OLAP 引擎的一個主力存儲。從 DWD 到 ClickHouse 的存儲只滿足了部分業(yè)務(wù)場景，還存在一些問題。比如我們需要做數(shù)據(jù)匯總，那么我們現(xiàn)在  DWS（Data Warehouse Service）層在這方面還稍微欠缺。目前明細(xì)數(shù)據(jù)進(jìn)入了  ClickHouse，我們首先對那些應(yīng)該匯總的數(shù)據(jù)存了明細(xì)，這樣會導(dǎo)致存儲量比較大，查詢效率較低。后續(xù)我們會考慮引入 Doris，因為它可以在實時計算側(cè)做實時聚合，依托 Doris 對 Update 的支持，就可以完善 DWS 功能。

這里展示的是我們的 SQL 編輯器。可以看到左邊是正在編輯的 SQL，我們支持  Flink 執(zhí)行計劃的查看、任務(wù)調(diào)試。右側(cè)一列可以定義源表、維表、輸出表。可以在自定義的數(shù)據(jù)源基礎(chǔ)上定義流表，并自動生產(chǎn) DDL。同時，對于某些自動生成 DDL 難以支持的場景，用戶可以在左邊的編輯區(qū)域自行編寫 DDL。

任務(wù)調(diào)式分為手動和自動兩種方式。手動方式需準(zhǔn)備樣例數(shù)據(jù)，拷貝到開發(fā)界面；自動方式則會從 SQL 任務(wù)的上游獲取樣例數(shù)據(jù)。元數(shù)據(jù)信息（kafka、HBase、ClickHouse 等）是動態(tài)獲得的，元信息和樣例共同生成的 DebugSQL 去調(diào)用 SQL 引擎的公共服務(wù)。SQL 引擎得到樣例數(shù)據(jù)后，比如，如果有關(guān)聯(lián)維表的操作，則會關(guān)聯(lián)線上維表，在 SQL 引擎中執(zhí)行調(diào)試，將結(jié)果送給 UI 端進(jìn)行展示。

這是一個完整的調(diào)試界面，可以看到左側(cè)是自動獲取的樣例數(shù)據(jù)，右側(cè)是下游的輸出。

根據(jù)元數(shù)據(jù)的定義及上報的指標(biāo)等監(jiān)控數(shù)據(jù)，我們可以生成一個實時數(shù)據(jù)血緣鏈路。圖中的箭頭展示了數(shù)據(jù)流轉(zhuǎn)的健康狀況，未來會對血緣鏈路上的數(shù)據(jù)監(jiān)控做得更細(xì)致。數(shù)據(jù)血緣滿足了 4 個方面的需求：溯源分析、問題排查、數(shù)據(jù)差異分析、提升用戶體驗。在血緣鏈路上還可以進(jìn)行比較復(fù)雜的異常預(yù)警，例如，數(shù)據(jù)源字段的變更對下游的影響。

這是我們 SQL2.0 引擎的大致架構(gòu)，通過 Antlr4 擴(kuò)展標(biāo)準(zhǔn) SQL 的語法，從而支持  Flink 的各種源，維表和下游存儲表的定義。通過 SqljobParser 內(nèi)置的 SqlStmtParser 生成 SqlContext，在邏輯計劃（Logical Plan）中做解析。如果遇到維表，則經(jīng)過一系列維表關(guān)聯(lián)的流程。上圖中下半部分是底層 API 架構(gòu)。

這是平臺 DDL 樣例。對于源表（Source），支持 Kafka，未來在新版本的 Flink 之上將可以支持更多種源。對于維表（Dim），支持 HBase、Redis、MySQL。數(shù)據(jù)存儲表（Sink）支持圖中所列五種。表格下面的是 DDL 定義的語法規(guī)則，右邊是一些表定義的樣例，分別是 Kafka 源表、維表和輸出表（輸出到控制臺）。

再看我們的維表關(guān)聯(lián)，從 SQL 引擎結(jié)構(gòu)可以看出，輸入的 SQL 進(jìn)行解析，當(dāng)有維表關(guān)聯(lián)時（包含 join 字段），我們會從語法層面做轉(zhuǎn)換。我們在表的層面定義了流和維關(guān)聯(lián)之后的表的形態(tài)，左下角是其生成過程。關(guān)聯(lián)維表、流維轉(zhuǎn)換、用異步 IO  獲取數(shù)據(jù)等過程不在這里細(xì)說。

隨著 Flink 社區(qū)新版本的發(fā)布，在 SQL 方面的支持越來越強(qiáng)，我們目前正在做基于 Flink1.11 的新版 SQL 引擎，也會將之前的 SQL 引擎統(tǒng)一。因為 Flink1.11 支持DDL，所以這部分我們不會再做，而是直接使用其新特性：

• 解析模塊（Parse Model）將用戶原始的 SQL 解析成內(nèi)部的執(zhí)行計劃，完全依賴于 Flink SQL。Connector Model 完成目前 Flink 尚未支持的 Connector 開發(fā)。

  
  

• Format Model 實現(xiàn)數(shù)據(jù)源字段的序列化和反序列化。

  
  

• 執(zhí)行模塊（Execute Model）基于 Flink1.11 SQL API 執(zhí)行解析后的執(zhí)行計劃。

  
  

• UDF 模塊是專門處理 UDF 的解析，如參數(shù)調(diào)用的合法驗證、權(quán)限驗證、細(xì)致的數(shù)據(jù)權(quán)限限制。

  
  

• SDK Model 是對外提供的標(biāo)準(zhǔn)化服務(wù)，如  SQL 文本開發(fā)的驗證，debug 功能等。

這是實時數(shù)倉的一個落地場景：交易的實時大屏，也是我們第一個落地的典型業(yè)務(wù)場景。我們支持各種交易實時指標(biāo)，用戶可以通過實時查詢 ClickHouse 得到交易數(shù)據(jù)的各種圖表展示。

客戶實時熱力圖是我們正在跟業(yè)務(wù)方溝通的一個需求場景，能實時獲取用戶線上的行為，使經(jīng)紀(jì)人對客戶行為有一個比較全面的實時掌控，促進(jìn)客戶維護(hù)的轉(zhuǎn)化率。另一方面，也使客戶更方便地了解房源熱度狀態(tài)，促使用戶做出購買決策。

四、事件驅(qū)動

先了解一下事件驅(qū)動型和數(shù)據(jù)分析型的區(qū)別：

• 事件驅(qū)動是根據(jù)事件流中的事件實時觸發(fā)外部計算和外部狀態(tài)的更新，主要關(guān)注實時事件觸發(fā)的外部變化，重在單獨事件以及外部動作的觸發(fā)。

  
  

• 數(shù)據(jù)分析型主要是從原始數(shù)據(jù)中提取有價值的信息，重在分析。

在我們跟業(yè)務(wù)方的溝通過程中，我們發(fā)現(xiàn)很多場景中他們希望實時獲取用戶的行為。比較典型的是風(fēng)控場景，根據(jù)用戶線上的行為模式判斷其是否觸發(fā)風(fēng)控規(guī)則。此外，我們的實時運(yùn)營，根據(jù)用戶線上行為給用戶進(jìn)行積分的增加及信息推送。搜索推薦也是我們非常關(guān)心的，即用戶在搜索之前的實時行為。綜合這些，我們提取出三方面問題：

• 一是用戶行為事件缺乏統(tǒng)一的抽象和管理，開發(fā)效率低，周期長，各部門存在重復(fù)建設(shè)；

  
  

• 二是規(guī)則邏輯與業(yè)務(wù)系統(tǒng)是耦合的，難以實現(xiàn)靈活的變化，對于復(fù)雜的規(guī)則或場景，業(yè)務(wù)方缺乏相關(guān)的技能和知識儲備，如對 CEP 的支持；

  
  

• 第三是缺乏統(tǒng)一的下游動作觸發(fā)的配置。

基于以上三個痛點，我們構(gòu)建了事件處理平臺，抽象成三個模塊，事件管理，規(guī)則引擎和動作觸發(fā)。

這是事件處理平臺所支持的業(yè)務(wù)場景。

這是事件處理平臺的架構(gòu)，總體來說就是管理模塊，引擎和動作觸發(fā)。在中間這里我們提供了一個適配層，可以跟第三方系統(tǒng)進(jìn)行集成。

這是我們事件處理的操作流程，首先是創(chuàng)建數(shù)據(jù)源，與實時計算平臺類似，主要支持 Kafka，在 Kafka 消息流上定義我們的數(shù)據(jù)格式。

在數(shù)據(jù)源基礎(chǔ)上創(chuàng)建事件流，事件流包含了同類事件，我們實現(xiàn)了一些算子，可以在數(shù)據(jù)源的基礎(chǔ)上做一些操作。從右側(cè)可以看到，在多個數(shù)據(jù)源上進(jìn)行了一些過濾、加解密的操作，最終通過 union 算子匯總成一個統(tǒng)一格式的同類事件的事件流，方便后續(xù)使用。

在事件流的基礎(chǔ)上可以定義單個的事件，之后可以創(chuàng)建事件組，以對接我們的業(yè)務(wù)含義，即明確具體的業(yè)務(wù)是做什么的，如用戶的點擊、瀏覽、分享、關(guān)注等事件。創(chuàng)建事件組有兩種方式：

• 一是本地方式，即可以根據(jù)事件的各個字段和維度設(shè)定條件；

  
  

• 二是遠(yuǎn)程方式，這與我們的埋點系統(tǒng)（用戶行為日志）直接連通，可以直接得到用戶事件的定義。

任務(wù)配置過程分幾個部分，這是 log 監(jiān)控的任務(wù)樣例。上圖展示的是事件處理的規(guī)則設(shè)置部分。這是一個 CEP 事件，可以定義事件窗口，獲取具體事件，在此之上定義 CEP 的模式，還可以定義事件的輸出，例如需要輸出哪些字段。

這是觸發(fā)動作調(diào)用，支持消息發(fā)送，服務(wù)調(diào)用及落地 Kafka。截圖展示的是消息發(fā)送的樣例。

五、未來規(guī)劃

這是我們實時計算的整體架構(gòu)，下部是 Hermes 實時計算平臺，主要包括任務(wù)管控、SQL 引擎、CEP 引擎等各種能力。Data Pipeline、實時數(shù)倉及事件處理平臺的任務(wù)都是通過此平臺進(jìn)行管控。未來我們計劃做的是用戶數(shù)據(jù)平臺，如各業(yè)務(wù)方對用戶的線上行為的歷史查詢，以及在全平臺用戶數(shù)據(jù)的綜合分析。

對未來的規(guī)劃主要有以上幾個方向，包括狀態(tài)的管理及恢復(fù)、動態(tài)的資源分配（動態(tài)的配置、動態(tài)的資源調(diào)整）。為了保持任務(wù)的穩(wěn)定性，我們在也計劃在高可用性方面做一些調(diào)研。在流批一體方面，會借用數(shù)據(jù)湖的能力，提供對歷史和實時數(shù)據(jù)的混合查詢的支持。

 

更多 Flink 相關(guān)技術(shù)交流，可掃碼加入社區(qū)釘釘大群～

---

▼ 關(guān)注「Flink 中文社區(qū)」，獲取更多技術(shù)干貨 ▼

  戳我，回顧作者分享視頻！