趣頭條基于Flink+ClickHouse的實時數(shù)據(jù)分析平臺
業(yè)務(wù)場景與現(xiàn)狀分析
Flink to Hive 的小時級場景
Flink to ClickHouse 的秒級場景
未來規(guī)劃
趣頭條的查詢頁面，分為離線查詢和實時查詢。離線查詢有 presto，spark，hive 等，實時查詢則引入了 ClickHouse 計算引擎。
上圖為實時數(shù)據(jù)報表，左邊為數(shù)據(jù)指標的曲線圖，右邊為詳細數(shù)據(jù)指標，目前數(shù)據(jù)指標的采集和計算，每五分鐘一個時間窗口，當然也會有三分鐘或者一分鐘的特殊情況。數(shù)據(jù)都是從 Kafka 實時導(dǎo)入 ClickHouse 進行計算的。
1. 小時級實現(xiàn)架構(gòu)圖
Flink-to-Hive 小時級實現(xiàn)架構(gòu)圖如圖所示，架構(gòu)實現(xiàn)的思路如下：
Database 中的 Binlog 抽數(shù)據(jù)到 Kafka，同時 Log server 數(shù)據(jù)也會上報到 Kafka，所有的實時數(shù)據(jù)落地到 Kafka 之后，通過 Flink 抽取到 HDFS 上。HDFS 到 Hive 之間有條虛線，即 Flink 落地到 HDFS 后，通過程序監(jiān)控，F(xiàn)link 在消費完成時，數(shù)據(jù)落地到 Hive 中可能是小時級的或者是半小時級的，甚至是分鐘級的，此時需要知道數(shù)據(jù)的 Event time 已經(jīng)到了什么時間，然后再去觸發(fā)比如 alert table、add partition、 add location 等，把分區(qū)寫進 Hive 中。這時還需要看一下當前的 Flink 任務(wù)的數(shù)據(jù)時間消費到了什么時間，如9點的數(shù)據(jù)要落地時，需要看一下 Kafka 里 Flink 數(shù)據(jù)消費是否到了9點，然后在 Hive 中觸發(fā)分區(qū)寫入。
2. 實現(xiàn)原理
這塊的實現(xiàn)原理主要是使用 Flink 高階版本的特性 StreamingFileSink。
StreamingFileSink 的主要功能如下：
forBulkFormat 支持 avro、parquet 格式，也就是支持鏈式的存儲格式
withBucketAssigner 自定義按數(shù)據(jù)時間分桶，支持數(shù)據(jù)時間的分桶，上圖用到該功能的地方定義了一個 EventtimeBucket，按照數(shù)據(jù)的時間落地到離線中
OnCheckpointRollingPolicy，會根據(jù) CheckPoint 時間來進行數(shù)據(jù)的落地，此處可以理解為按照數(shù)據(jù)的時間，比如按照一定的 CheckPoint 時間內(nèi)進行數(shù)據(jù)落地、回滾，數(shù)據(jù)落地策略還可以按照數(shù)據(jù)大小落地
Exactly-Once 語義實現(xiàn)，F(xiàn)link 中自帶的 StreamingFileSink 是用 Exactly-Once 語義來實現(xiàn)的。Flink 中有兩個 Exactly-Once 的實現(xiàn)，第一個是 Kafka 的 Exactly-Once，第二個是 StreamingFileSink 實現(xiàn)了 Exactly-Once 語義，像上圖中 CheckpointRollingPolicy 設(shè)置的是十分鐘落地一次到 HDFS 文件中
下面來具體說一下 Exactly-Once 是如何實現(xiàn)的。
① Exactly-Once
具體實現(xiàn) Exactly-Once 的方式，如上圖所示，左側(cè)是一個二階段的模型，Coordinator 發(fā)一個 perpare，所有的參與者或者執(zhí)行者開始觸發(fā) ack 動作，Coordinator 收到所有人的 ack 動作后，就開始執(zhí)行 commit，所有的執(zhí)行者就把左右的數(shù)據(jù)進行落地。到了 Flink 這塊，Source 收到了 checkpoint barrier 流的時候，開始觸發(fā) snapshorState 發(fā)送到 Job Manager，Job Manager 把所有的 CheckPoint 都完成以后，會發(fā)送一個 notifyCheckpointComplete，F(xiàn)link 這塊跟上圖左邊的二階段提交協(xié)議是一致的，F(xiàn)link 也是可以實現(xiàn)二階段提交協(xié)議的。
② 如何使用 Flink 實現(xiàn)二階段提交協(xié)議
首先 StramingFileSink 實現(xiàn)了兩個接口，分別是 CheckpointedFunction 和 CheckpointListener。
CheckpointedFunction 實現(xiàn)了 initialzeState 和 snaoshotState 這兩個函數(shù)；
CheckpointListener 是 notifyCheckPoint Complete 的方法實現(xiàn)。
所以這兩個接口可以實現(xiàn)二階段提交的語義，initialzeState 算子剛啟動的時候，它會啟動三個動作 commitpendingFile、restoreInProgressFile、truncate。
第一步 commitpedingFile，也就是實時的數(shù)據(jù)落地到 HDFS 的時候，有三個狀態(tài)，第一個狀態(tài)是 in-progress，即正在進行中的一個狀態(tài)，第二個狀態(tài)是 pending 的狀態(tài)，第三個狀態(tài)是 finish 的狀態(tài)。
在實時的寫入時，如果 CheckPoint 還沒有在這之間成功的時候，程序出問題了，那接下來啟動的時候就會觸發(fā) initialzeState，會把曾經(jīng) pending 的 file 進行 commit，然后把寫了一半的文件比如 in-progress 文件重置或者截斷，進行重置或者截斷是使用的是 Hadoop 的2.7版本的 turncate 方式。也就是數(shù)據(jù)在一直寫入，但是寫入沒有達到一個 CheckPoint 周期，也就是說中間數(shù)據(jù)斷開了，下一次啟動的時候，要么把之前沒有寫完整的數(shù)據(jù)截斷掉，之前 CheckPoint 觸發(fā)已經(jīng)寫好的數(shù)據(jù)直接 commit。
第二步 invoke 就是數(shù)據(jù)實時的寫入
第三步 snapshotState 在觸發(fā) CheckPoint 的時候會把 in-progress 文件轉(zhuǎn)成 pending state 文件，也就是開始提交文件，同時記錄 length 長度。記錄長度是因為前邊的步驟需要 truncate 來截斷多長，snapshot 時，是沒有真正的寫入到 HDFS，其實是寫入到 ListState，等所有的 CheckPoint 算子都完成了，就把 ListState 中的數(shù)據(jù)都刷到 HDFS 中，只要數(shù)據(jù)存在 Flink 自帶的 state 中，不斷把數(shù)據(jù)成功的刷到 HDFS 中就行了。
第四步 notifyCheckPoint Complete 會觸發(fā) pending 動作到 finished 狀態(tài)的數(shù)據(jù)寫入，實現(xiàn)的方式直接使用 rename，Streaming 會不斷的寫入 HDFS 中的臨時文件，等到 notifyCheckPoint 結(jié)束之后，直接做一個 rename 動作，寫成正式文件。
3. 跨集群多 nameservices
趣頭條的實時集群跟離線集群是獨立的，實時集群目前是一套，離線集群是有多套。通過實時集群要寫入到離線集群，這樣就會遇到一個問題，HDFS nameservices 問題，如果在實時集群中把所有的離線集群的 nameservice 用 namenode HA 的方式全部打入到實時集群，是不太合適的。所以使用 Flink 任務(wù)中 resource 下邊把 HDFS 中的 xml 文件中間加 final 標簽，設(shè)置為 true。此處的 value 標簽中，stream 是一個實時集群，date 是一個離線集群，這樣把兩個 HA 配置在 value 標簽，從而達到實時集群是實時集群，離線集群是離線集群，中間的 HDFS 中 set 不需要相互修改，直接在客戶端時間就行了。
4. 多用戶寫入權(quán)限
針對多用戶權(quán)限寫入的問題，實時寫入離線 HDFS 中的時候，會涉及到用戶權(quán)限。遇到用戶權(quán)限時，也會有一個問題，F(xiàn)link 實時提交的用戶，是定義好的，所有的程序里用戶是同一個，但是離線是多個用戶，F(xiàn)link 目前對于這塊用戶的權(quán)限做的還不夠好，所以我們自己改造了一下，在 API 中添加了 withBucketUser，上邊已經(jīng)配置好了 nameServices，然后通過該參數(shù)來配置具體是那個用戶來寫入 HDFS 中，這是 API 層級的。
API 層級的好處是一個 Flink 程序可以寫多個，可以指定不同的 HDFS 的不同的用戶就可以。具體實現(xiàn)就是在 Hadoop file system 中加一個 ugi.do as，代理用戶。以上是趣頭條用 Flink 在實時數(shù)據(jù)同步到 Hive 做的一些工作。其中會有一些小文件的問題，針對小文件，我們通過后臺程序定期的 merge，如果 CheckPoint 的時間很短，就會出現(xiàn)大量的小文件的問題。
1. 秒級實現(xiàn)架構(gòu)圖
首先來解釋一下趣頭條使用 Flink+ClickHouse 的場景，最開始展示的很多實時指標，可能是每五分鐘計算一次，也可能是每三分鐘計算一次。如果每一個實時指標用一個 Flink 任務(wù)，即使是 FlinkSQL 來寫，比如消費一個 Kafka Topic，計算它的日活、新增、流程等，當用戶提出一個新的需求，那這個 Flink 任務(wù)是需要修改還是再啟動一個 Flink 任務(wù)來消費這個 Topic，這樣的話就會出現(xiàn) Flink 任務(wù)在不斷的修改或者不斷的啟動新的 Flink 新的任務(wù)。為了解決這個問題，就讓 Flink 后邊接一個套 ClickHouse 實現(xiàn)整體的 OLAP。
上圖為秒級實現(xiàn)架構(gòu)圖，從 Kafka 到 Flink 到 Hive 然后再到 ClickHouse 集群，對接外部 Horizon ( 實時報表 )、QE ( 實時 adhoc 查詢 )、千尋 ( 數(shù)據(jù)分析 )、用戶畫像 ( 實時的用戶畫像 )。
2. Why Flink+ClickHouse
具體來說為什么要用 Flink+ClickHouse，主要有以下幾點：
指標實現(xiàn)支持 sql 描述，以前的方案使用是 storm 的程序，通過 stormsql 實現(xiàn)，包括 flinksql，這些內(nèi)容對于 UDF 支持相對有限，但是現(xiàn)在這套 Flink+ClickHouse 基本上可以把分析師提的指標通過 sql 實現(xiàn)。
指標的上下線互不影響，這個主要是解決上邊提到的關(guān)于 Flink 任務(wù)消費了 topic 以后，假如用戶提出新的指標的時候，是啟動新任務(wù)還是要不斷修改的問題。
數(shù)據(jù)可回溯，方便異常排查，這個就類似上邊提到的假如我的日活掉了，需要知道哪些指標的口徑的邏輯掉了、哪個上報的數(shù)據(jù)掉了，如 cmd 掉了還是數(shù)據(jù)流 kafka 掉了還是用戶上報的時候指標沒有上報導(dǎo)致的日活掉了。假如單純的 flink 的話，只是會計算出那個指標掉了，是沒辦法回溯的。
計算快，一個周期內(nèi)完成所有的指標計算，現(xiàn)在的 horizon 曲線可能是幾百上千，需要在五分鐘之內(nèi)或者十分鐘之內(nèi)，把所有分時、累時、以及維度下降的指標全部計算出來。
支持實時流，分部署部署，運維簡單。
目前趣頭條 Flink 集群有 100+ 臺 32 核 128 G 3.5T SSD，日數(shù)據(jù)量 2000+ 億，日查詢量 21w+ 次，80% 查詢在 1s 內(nèi)完成。
上圖為單表測試結(jié)果。ClickHouse 單表測試速度快。但受制于架構(gòu)，ClickHouse 的 Join 較弱。
上圖是處理相對較為復(fù)雜的 SQL，count+group by+order by，ClickHouse 在 3.6s內(nèi)完成 26 億數(shù)據(jù)計算。
3. Why ClickHouse so Fast
接下來說一下為什么 ClickHouse 這么快，主要是有以下幾點：
列式存儲+LZ4、ZSTD 數(shù)據(jù)壓縮：列式存儲基本是通用的。
計算存儲本地化+向量化執(zhí)行：計算存儲本地化，ClickHouse 跟 presto 不一樣，presto 數(shù)據(jù)可能存在 Hadoop 集群里邊或者 HDFS 中，需要把數(shù)據(jù)拉過來，然后進行實時的計算；而 ClickHouse 是每一臺計算機器需要的數(shù)據(jù)存儲在本地的 ssd 盤，只要計算本地的數(shù)據(jù)就可以了，比如求 count 之類的，計算完成后把其他的節(jié)點進行合并就可以了。
LSM merge tree+Index：LSM merge tree，他會不斷的使用 batch 的形式把數(shù)據(jù)寫入到 ClickHouse 之后，在后臺做了一個線程把數(shù)據(jù)進行 merge，做一個 index 索引，也就是給這張數(shù)據(jù)表建立很多索引，類如常見的 DT 的時間索引、小時級的數(shù)據(jù)索引來提高查詢性能或者速度。
SIMD+LLVM 優(yōu)化：SIMD 就是一個單指令多數(shù)據(jù)集，LLVM 是一個 C++ 的編譯器
SQL 語法、UDF 完善：在這塊有很大的需求，比如數(shù)據(jù)分析以及維度下墜，常規(guī)的 horizon 數(shù)據(jù)報表可能就是 count、sum、以及 group by、order by 等，但是在一些維度下墜或者是數(shù)據(jù)分析領(lǐng)域，可能會有一個窗口期的概念，在一段窗口期內(nèi)的留存，所以要用到一些更高的特性，類如時間窗口的功能。
上圖是 MergeTree 的運行原理圖解，最上邊的第一層是數(shù)據(jù)一個 batch 一個 batch 的實時寫入，后臺會做每一個層級的數(shù)據(jù) merge，這塊跟 HBase 差不多的實現(xiàn)，merge 的時候會進行數(shù)據(jù)的排序，然后做一個數(shù)據(jù)索引。
上圖是 ClickHouse Connector，ClickHouse 有兩個概念，local table 和 distribute table。local table 是用來寫的，當然 distribute table 也可以寫入，但是會出現(xiàn)很大的 io 問題，所以盡量不要寫 distribute table。但是可以讀 distribute table。5-10w 一個 batch 進行數(shù)據(jù)寫入，正常的情況下，是5秒一個周期。
RoundRobinClickHouse DataSource 這塊是趣頭條自己實現(xiàn)的；
ClickHouse 官方 API 使用：
BalancedClickHouseDataSource 實現(xiàn)的。
上圖是 ClickHouse 官方 API 使用：
BalancedClickHouseDataSource
里邊有一個問題，比如 mysql 配置一個 ip 和端口號就可以把數(shù)據(jù)寫入了，但是這塊要寫入 local table 的，所以必須要知道這一個集群到底有多少 local table，每一個 local table 的 ip 和端口號，假如有100臺機器，就必須要把這100臺機器的 ip 和端口號配置好，然后進行寫入。
官方的 api 中有兩個 schedule：
一個是 scheduleActualization
另一個是 scheduleConnectionsCleaning
第一個是指100臺機器配置了100個 ip 或者端口號，可能會有一些機器出現(xiàn) ping 不通或者服務(wù)無響應(yīng)，這塊是定時的做一個 Actualiza 來發(fā)現(xiàn)這些機器哪些無法連接，觸發(fā)一個下限來把這些 ip 刪除掉。
第二個 scheduleConnectionsCleaning，因為 ClickHouse 是 http 的方式，定期的會把一些沒用的 http 的請求清理掉。
針對于官方提供的 API，趣頭條對這方面做了一個加強，開發(fā)了一個 RoundRobinClickHouseDataSource，實現(xiàn)了三個語義，分別是 testOnBorrow、testOnReturn、testWhileldle。
第一個 testOnBorrow 取鏈接的時候，設(shè)置 為true，然后去 ping 一下這個鏈接能不能拿到，ClickHouse 寫入的時候，使用的 batch，所以盡量就是拿鏈接的時候要拿到成功的鏈接；第二個 testOnReturn 設(shè)置為 false，testWhileldle 設(shè)置為 true，把上邊官方的兩個 schedule 功能集成進去了。為什么要實現(xiàn) RoundRobin，主要是因為假如有100臺機器，ClickHouse 相對于 Hadoop 來說，還是需要好好維護一下，如果是 insert 的話，后臺是不斷 merge 的過程，insert 速度大于 merge 速度時候，會導(dǎo)致 merge 速度永遠跟不上，所以就寫完這臺機器接下來寫別的機器，以及5秒一個間隔的寫，使 merge 的速度盡量跟上 insert 的速度，這塊是整個部分最需要注意的地方。
4. Backfill
趣頭條針對集群容錯做了一些優(yōu)化，主要包括兩點：
第一點是 Flink 任務(wù)小時級容錯
第二點是 ClickHouse 集群小時級容錯
Flink 導(dǎo)入數(shù)據(jù)到 ClickHouse，來實現(xiàn)數(shù)據(jù)的查詢、報表展示，會遇到一些問題。如 Flink 任務(wù)出現(xiàn)故障、報錯、數(shù)據(jù)反壓、network 的一些問題；或者 ClickHouse 集群出現(xiàn)了一些不可響應(yīng)、ZK 跟不上等 ZK 問題；或者集群的負載問題；或者是上邊提到的 insert 太快的問題；會導(dǎo)致整個任務(wù)都有問題。如果數(shù)據(jù)量突然暴漲，把 Flink 啟動，就會出現(xiàn)一段時間內(nèi)不停的追數(shù)據(jù)，可能就需要調(diào)大它的并行度之類的，讓 Flink 任務(wù)把數(shù)據(jù)追上。但是數(shù)據(jù)已經(jīng)積壓了，F(xiàn)link 又要加大它的并發(fā)度來處理數(shù)據(jù)，但是 ClickHouse 那塊又限制了 insert 速度不能太快，所以就做了另外一個機制，也就是 Flink 故障了或者 ClickHouse 故障了，等到 ClickHouse 集群恢復(fù)之后，F(xiàn)link 任務(wù)還是從最新的開始消費，過去的一段數(shù)據(jù)不再去追了，通過 Hive 來把數(shù)據(jù)導(dǎo)入到 ClickHouse。
用 Hive 是因為數(shù)據(jù)通過 Kafka 已經(jīng)實時落地到 Hive，通過 waterdrop 把數(shù)據(jù)寫入到 ClickHouse，ClickHouse 是有分區(qū)的，只要把上一個小時的數(shù)據(jù)刪除，再把 Hive 一個小時的數(shù)據(jù)導(dǎo)入進來，這樣就可以繼續(xù)提供數(shù)據(jù)查詢操作了。
最后是對未來的發(fā)展與思考。
1. Connector SQL
對于未來的發(fā)展，首先是 Connectors SQL，也就是把 Connector 進行 SQL 化，現(xiàn)在是 Flink-to-Hive 以及 Flink-to-ClickHouse，相對來講，都是比較固化的一些場景，所以是可以進行 sql 化，除了把 HDFS 的路徑指定以及用戶指定，其他的一些過程都是可以 SQL 化描述出來的。
2. Delta lake
Flink 是流批一體計算引擎，但是沒有流批一體的存儲。趣頭條會用 HBase、Kudu、Redis 等能夠與 Flink 實時交互的 KV 存儲進行數(shù)據(jù)計算。如計算新增問題，目前趣頭條的方案是需要將 Hive 歷史用戶刷到 Redis 或 HBase 中，與 Flink 進行實時交互判斷用戶是否新增。但因為 Hive 中的數(shù)據(jù)和 Redis 中的數(shù)據(jù)是存儲為兩份數(shù)據(jù)。其次 Binlog 抽取數(shù)據(jù)會涉及 delete 動作，Hbase，Kudu 支持數(shù)據(jù)修改，定期回到 Hive 中。帶來的問題是 HBase，Kudu 中存在數(shù)據(jù)，Hive 又保存了一份數(shù)據(jù)，多出一份或多份數(shù)據(jù)。如果有流批一體的存儲支持上述場景，當 Flink 任務(wù)過來，可以與離線數(shù)據(jù)進行實時交互，包括實時查詢 Hive 數(shù)據(jù)等，可以實時判斷用戶是否新增，對數(shù)據(jù)進行實時修改、更新或 delete，也能支持 Hive 的批的動作存儲。未來，趣頭條考慮對 Flink 做流批的存儲，使 Flink 生態(tài)統(tǒng)一為流批結(jié)合。
嘉賓介紹：
王金海，10 年互聯(lián)網(wǎng)歷練，先后在唯品會負責(zé)用戶畫像系統(tǒng)，提供人群的個性化營銷服務(wù)；餓了么擔任架構(gòu)師，負責(zé)大數(shù)據(jù)任務(wù)調(diào)度、元數(shù)據(jù)開發(fā)、任務(wù)畫像等工作；現(xiàn)為趣頭條數(shù)據(jù)中心平臺負責(zé)人，負責(zé)大數(shù)據(jù)基礎(chǔ)計算層 ( spark、presto、flink、clickhouse )、平臺服務(wù)層 ( libra 實時計算、kepler 離線調(diào)度 )、數(shù)據(jù)產(chǎn)品層 ( qe即時查詢、horizon 數(shù)據(jù)報表、metadata 元數(shù)據(jù)、數(shù)據(jù)權(quán)限等 )、以及團隊建設(shè)。
特別推薦一個分享架構(gòu)+算法的優(yōu)質(zhì)內(nèi)容，還沒關(guān)注的小伙伴，可以長按關(guān)注一下：
長按訂閱更多精彩▼
serif;letter-spacing: 0.544px;white-space: normal;text-align: right;line-height: 2em;box-sizing: border-box !important;word-wrap: break-word !important;">如有收獲，點個在看，誠摯感謝
免責(zé)聲明：本文內(nèi)容由21ic獲得授權(quán)后發(fā)布，版權(quán)歸原作者所有，本平臺僅提供信息存儲服務(wù)。文章僅代表作者個人觀點，不代表本平臺立場，如有問題，請聯(lián)系我們，謝謝！
ckquote>
下載文檔
