作者：王剛、劉首維
在 2019 年之前，之家的大部分實時業務都是運作在 Storm 之上的。Storm 作為早期主流的實時計算引擎，憑借簡單的 Spout 和 Bolt 程式設計模型以及叢集本身的穩定性，俘獲了大批使用者。下圖是實時計算團隊 Storm 平台頁面：
自 2015 年至今 Storm 在之家已經運作 4 年之久，但随着實時計算的需求日漸增多，資料規模逐漸增大，Storm 在開發及維護成本上都凸顯了不足，這裡列舉兩個痛點:
1.翻譯 SQL
我們一直是 Lambda 架構，會用 T+1 的離線資料修正實時資料，即最終以離線資料為準，是以計算口徑實時要和離線完全保持一緻，實時資料開發的需求文檔就是離線的 SQL，實時開發人員的核心工作就是把離線的 SQL 翻譯成 Storm 代碼，期間雖然封裝了一些通用的 bolt 來簡化開發，但把離線動辄幾百行的 SQL 精準的翻譯成代碼還是很有挑戰的，并且每次運作都要經過打包，上傳， 重新開機的一系列的繁瑣操作，調試成本很高。
2.過于依賴外部存儲
Storm 對狀态支援的不好，通常需要借助 Redis，HBase 這類 kv 存儲維護中間狀态，我們之前是強依賴 Redis。比如常見的計算 UV 的場景， 最簡單的辦法是使用 Redis 的 sadd 指令判斷 uid 是否為已經存在，但這種方法會占用大量記憶體，如果沒有提前報備的大促或搞活動導緻流量翻倍的情況，很容易把 Redis 記憶體搞滿，運維同學也會被殺個措手不及，同時 Redis 的吞吐能力也限制了整個作業的吞吐量。
在此背景下我們封裝了基于 BloomFilter 的 bolt，BloomFilter 本身也會作為狀态定期持久化到 reids 中，但是在多元度高基數的場景下，很難精确控制每個 BloomFilter 的大小，同樣會占用很大記憶體。同時，過于依賴 Redis，在 Redis 叢集 rtt 過長或部分節點負載高時會導緻 Storm 作業 failed。
我們從 2018 年開始調研 Flink 引擎，其相對完備的 SQL 支援，天生對狀态的支援吸引了我們，在經過學習調研後，2019 年初開始設計開發 Flink SQL 平台，目前平台已經服務于數倉、監控、日志、運維、測試等團隊，2019 年 10 月已經有 160+ 線上作業，每日計算量 5000 億條 支援實時數倉，實時推薦，UAS 系統，日志看闆，性能測試等多種場景。單任務目前最高為 200 萬 QPS。平台能夠得到快速廣泛的應用，主要得益以下幾點:
- 開發成本低：之家大部分的實時任務可以用 Flink SQL + UDF 實作。平台提供常用的 Source 和 Sink，以及業務開發常用的 UDF，同時使用者可以自己編寫 UDF。基于“ SQL + 配置”的方式完成開發，可以滿足大部分需求。對于自定義任務，我們提供友善開發使用的 SDK，助力使用者快速開發自定義 Flink 任務。平台面向的使用者已經不隻是專業的資料開發人員了，普通開發、 測試、運維人員經過簡單教育訓練都可以在平台上完成日常的實時資料開發工作，實作平台賦能化。
- 高性能：Flink 可以完全基于狀态(記憶體，磁盤)做計算，對比之前依賴外部存儲做計算的場景，性能提升巨。在 818 活動壓測期間，改造後的程式可以輕松支援原來幾十倍流量實時計算流量，且橫向擴充性能十分良好。
- 維護成本低：使用者将任務托管在平台上，任務的存續由平台負責，使用者專注于任務本身的邏輯開發本身即可。對于 SQL 任務，SQL 的可讀性極高，便于維護；對于自定義任務，基于我們 SDK 開發，使用者可以更專注于梳理業務邏輯上。不論是 SQL 任務還是 SDK，我們都内嵌了大量監控，并與報警平台關聯，友善使用者快速發現分析定位并修複任務，提高穩定性。
- 支援數倉分層模型：平台提供了良好的 SQL 支援，數倉人員可以借助 SQL，将離線數倉的建設經驗應用于實時數倉的建設上。
- 資料資産管理：SQL 語句本身是結構化的，我們通過解析一個作業的 SQL，結合 source、 sink 的 DDL，可以很容易的知道這個作業的上下遊，天然保留血緣關系。
下面将分三部分給大家分享:
- 架構及設計思路
- 基于 Flink SQL 平台的實時數倉的實踐及使用案例
- 後續規劃
一. 架構及設計思路
1.表管理
在平台上我們把 source，sink 都抽象成表:
表管理: 目前我們是基于 Flink 1.7.2 的，這個版本還不支援 DDL，是以我們通過擴充 Calcite 文法，自己實作了 DDL 解析，把 source 和 sink 階段使用的外部存儲 (Kafka、Mysql 、ES、Redis、Http 等)中的目标對象都映射成關系型表管理起來，友善複用。如下圖:
對于動态配置，可以在每個任務界面的配置功能靈活地進行指定。
血緣關系：每次運作任務我們都會解析出這個任務需要的源表(流表，維表)和結果表 ，可以很自然的儲存各種表的血緣關系。
2.任務配置管理
2.1 SQL
我們 SQL 任務支援兩種模式:一種是直接 SELECT 查詢，一種是 INSERT INTO 将資料寫入外部存儲的目标表 。
- 使用者執行 SELECT 的時候，我們會在頁面上滾動展示計算結果，以供檢視結果及調試。
- SELECT 實作思路是 Flink 的計算結果會發送緩存伺服器，并存儲在 Ringbuffer 中。
- 同時，在同一個作業中可以編寫多個 SQL，我們還提供了建立視圖的 DDL，簡化對複雜 SQL 的多次引用。
2.2 任務配置
任務配置有三種 :
- 作業配置(jobConf) : 如 checkpoint 時間間隔，狀态過期時間，重新開機政策等等。
- 啟動配置 (launchConf) : 如 jobmanager 記憶體，taskmanger 個數，slot 數 ，使用的隊列等資源相關的配置及執行代碼的版本。
- 叢集配置(clusterConf) : 因為我們的任務模式是 perjob 的，即每個 Flink 作業都運作在一個獨立的 Flink 叢集中，我們可以很輕松的給每個作業配置個性化的叢集參數。
3.權限管理
我們的平台是支援多租戶的，目前在以下兩方面做了權限控制:
- 作業級别：每個團隊隻能去管理自己團隊内的作業。
- 表級别：平台上的表分兩類，一類是團隊内部的表，僅限于團隊内部可見;另一類是公共表，對所有的使用者可見，公共表中的流表(Kafka topic) 目前需要去消息平台去申請讀/寫權限才能使用。
4.UDF 管理
目前所有啟動 Flink 作業的請求都是通過一個 client 程序去送出的，因為是同一個程序，是以不能做到頻繁的加載 jar 包，導緻目前還不能自助上傳 UDF 。 我們接下來準備參考 athenax 的做法，在每次去運作任務的時候單獨起個程序去編譯 Jobgraph 對象再轉發給送出程序，這樣可以做到團隊間的 UDF 不沖突。
5.資源排程架構
資源排程架構我們使用的是 YARN 和 k8s 狀态存儲在 hdfs 上。之家 Hadoop 伺服器叢集數量在幾千台左右，并且在之家的業務場景下實時和離線計算是天生錯峰的，白天流量高，離線的作業量少，YARN 計算資源充足，完全可以滿足實時計算的需求。
而 k8s 對我們來說是未來發展的方向，有更高的穩定性和更友善的使用體驗。 目前我們有少量作業運作在 k8s 上，接下來我們在考慮在 YARN 叢集和 k8s 或者備用 YARN 叢集之間做熱備，比如 YARN 叢集 down 掉重要的作業會從 hdfs 讀取狀态切到 k8s 上。
6.日志收集
我們首先定制了自己的 Log4j Layout 增加了輔助的日志資訊。對于運作在 YARN 上的任務, 我們基于 Flume 的 Log4j Appender 定制了自己的日志收集器，從伺服器異步發送日志到 Kafka 中，盡可能地降低對運作作業的影響；對于運作在 k8s 的任務，我們通過容器組同僚提供的 API 直接采集即可。
日志會上報到公司統一的采集系統，途經 Kafka 最終寫入 Elasticsearch 叢集，通過 Kibana 可以友善的檢視 Flink 日志 。
7.監控報警
監控報警這塊主要依賴公司的統一監控平台。
- 統一監控平台提供了專用的 Promethues 服務，我們通過 Flink 的 promethues push gateway 上報 metric ，Flink 本身的 metric 比較完善，其監控的粒度精确到每個 subtask 的 operator。
- 我們自行開發的元件都定義了常用的 metric，使用者可以通過檢視監控圖表定位元件的健康狀态。
- 我們配置了一些預設的模版，包括 Flink Cluster, IO(task/operator 級别), JVM 、Kafka source、Elasticsearch sink、Redis sink 等，友善使用者檢視圖表，在統一監控平台上，使用者可以針對任意 metric 設定自定義報警。
- 同時對延遲，任務重新開機，等重要名額提供了預設的報警機制。
Kafka-connector監控
Redis-connector監控
二.基于 Flink 平台的實時數倉的實踐
如上文提到，我們把 Kafka 的 topic 當做 table 并結合消息平台做權限控制。這樣做的目的就是友善給做實時數倉做準備。資料源來自 mysql 的 binlog 日志、埋點的流量日志、及伺服器系統日志。
其中在資料表打寬的過程我們借鑒了袋鼠雲的維表 join 辦法，用 calcite 再解析成 SqlNode 之後就把表合并成一張寬表，再修改原始的 SQL 語句。目前被用在表的清洗打寬還是沒有問題的，但是這種方式在很複雜的 SQL 語句裡就沒那麼好用了，隻能借助 udtf。
我們把資料清洗成寬表，再基于寬表清洗成每個主題的彙總表，最終将彙總表和寬表開放給業務方使用，業務方可以直接在平台通過編寫 SQL 完成實時計算的開發工作。
目前已經支援了推送，推薦，數倉，UAS，監控，日志，壓測，羅盤大屏等多種業務場景。
1.使用案例：推薦系統實時名額計算
内容和資訊一直是之家的核心和根本。内容推薦系統更是支撐整個内容咨詢體系的一個重要組成部分。基于實時計算平台的 SQL 子產品計算了大量次元的實時名額，寫入多種存儲，幫助推薦系統快速而準确地回報推薦物料的推薦效果，形成有效的閉環。
之家對于每個投放的物料/資源，都有唯一的業務類型(biztype)和物料 Id (objectid),通過這兩個屬性可以唯一表示一個資源。對于針對使用者的不同的操 作和行為，我們定義了 eventid 這個概念，比如可見曝光，點選等。而從實驗/ 政策的次元上，我們存在實驗主題(topic)和分桶 id(bucketid)的概念。此外， 次元還包括但不限于地域，IP。另一方面，還以檢測資源投放耗時和響應時間等名額。
衆所周知，一個 Flink 的計算任務分為三個階段 source -> transformation -> sink，那我們也結合 SQL 來看這三個階段是如何完成的。
- source
使用者行為日志(UAS)是寫入 Kafka 的，格式是 JSON。我們首先做的事情是對原始日志進行清洗和整理抽象成關系型模型。利用平台的資料集解析建立功能搭配幾 個 SQL 清洗資料任務還是比較容易的:
分析資料格式，利用平台建表功能建立對于産出的一系列關系型表的 Schema。
利用 SQL 子產品編寫清洗資料任務，對應步驟 1 的目标産出表。
- kafkasource
sink 和 source 的步驟差不多，平台目前支援多種 sink 的同時，還基于 javacc 提供靈活的自定義寫入模闆，使用者可以自己決定資料是怎樣組織寫入 sink 的，可以覆寫絕大部分需求，如果遇到不能滿足的情況，也可以通過繼承我們對外釋出的 SDK 的接口完成自定義邏輯的編寫。
transformation 相對于另外兩個階段，複雜在業務邏輯上，下面舉例說明利用 SQL 快速解決問題。
- 需求：計算單一物料每天的可見曝光和點選的 pv 和 uv。
- 開發：定義好 sink，直接編寫 SQL 開發即可。
2.總結
利用實時平台，我們很好地将推薦的實時名額計算的系列任務，分層化、子產品化、規範化，開發速度與準确性大幅提升，最快一個新的名額計算隻需要小時級就可以搞定，同時學習成本大幅降低，使用者隻需要使用我們提供的頁面+ SQL 就可以完成實時任務的開發，進而賦能業務方，使之可以獨立開發實時計算任務。
同時也解決了我們之前提到過的兩個痛點:
不需要把離線的 SQL 翻譯成代碼，基于清洗過的寬表，直接用 SQL 就可以實作實時名額的開發。
不再重度依賴第三方存儲存放狀态，Flink 自身維護了狀态，Redis 隻是單純的存儲最終結果。
三.後續規劃
1.與倉庫，業務方合作生産更多的業務寬表，彙總表，将資料資産化。
2.不斷豐富平台功能，支援更多 Sink 與 Source，提供更多的工具供業務方使用，進一步降低開發運維成本。
3.将平台任務部署繼續向 K8s 模式傾斜。
4.持續不斷提升 Flink 在公司的影響，吸納更多人使用 Flink 解決生産問題，豐富使用場景。
5.調研 Flink 1.9 以後的版本，并逐漸引入到公司生産中。
**▼ Apache Flink 社群推薦 ▼
**
Apache Flink 及大資料領域盛會 Flink Forward Asia 2019 将于 11月28-30日在北京舉辦，阿裡、騰訊、美團、位元組跳動、百度、英特爾、DellEMC、Lyft、Netflix 及 Flink 創始團隊等近 30 家知名企業資深技術專家齊聚國際會議中心，與全球開發者共同探讨大資料時代核心技術與開源生态。了解更多精彩議程請點選：
https://developer.aliyun.com/special/ffa2019-conference?spm=a2c6h.13239638.0.0.21f27955RBnbkV