<section style="margin-bottom: 5px;max-width: 100%;font-family: -apple-system-font, BlinkMacSystemFont, "Helvetica Neue", "PingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;letter-spacing: 0.544px;white-space: normal;background-color: rgb(255, 255, 255);text-align: center;box-sizing: border-box !important;overflow-wrap: break-word !important;">

導(dǎo)讀： 趣頭條一直致力于使用大數(shù)據(jù)分析指導(dǎo)業(yè)務(wù)發(fā)展。目前在實(shí)時(shí)化領(lǐng)域主要使用 Flink+ClickHouse 解決方案，覆蓋場(chǎng)景包括實(shí)時(shí)數(shù)據(jù)報(bào)表、Adhoc 即時(shí)查詢、事件分析、漏斗分析、留存分析等精細(xì)化運(yùn)營(yíng)策略，整體響應(yīng) 80% 在 1 秒內(nèi)完成，大大提升了用戶實(shí)時(shí)取數(shù)體驗(yàn)，推動(dòng)業(yè)務(wù)更快迭代發(fā)展。

本次分享主要內(nèi)容：

• 業(yè)務(wù)場(chǎng)景與現(xiàn)狀分析

• Flink to Hive 的小時(shí)級(jí)場(chǎng)景

• Flink to ClickHouse 的秒級(jí)場(chǎng)景

• 未來(lái)規(guī)劃

趣頭條的查詢頁(yè)面，分為離線查詢和實(shí)時(shí)查詢。離線查詢有 presto，spark，hive 等，實(shí)時(shí)查詢則引入了 ClickHouse 計(jì)算引擎。

上圖為實(shí)時(shí)數(shù)據(jù)報(bào)表，左邊為數(shù)據(jù)指標(biāo)的曲線圖，右邊為詳細(xì)數(shù)據(jù)指標(biāo)，目前數(shù)據(jù)指標(biāo)的采集和計(jì)算，每五分鐘一個(gè)時(shí)間窗口，當(dāng)然也會(huì)有三分鐘或者一分鐘的特殊情況。數(shù)據(jù)都是從 Kafka 實(shí)時(shí)導(dǎo)入 ClickHouse 進(jìn)行計(jì)算的。

1. 小時(shí)級(jí)實(shí)現(xiàn)架構(gòu)圖

Flink-to-Hive 小時(shí)級(jí)實(shí)現(xiàn)架構(gòu)圖如圖所示，架構(gòu)實(shí)現(xiàn)的思路如下：

Database 中的 Binlog 抽數(shù)據(jù)到 Kafka，同時(shí) Log server 數(shù)據(jù)也會(huì)上報(bào)到 Kafka，所有的實(shí)時(shí)數(shù)據(jù)落地到 Kafka 之后，通過(guò) Flink 抽取到 HDFS 上。HDFS 到 Hive 之間有條虛線，即 Flink 落地到 HDFS 后，通過(guò)程序監(jiān)控，F(xiàn)link 在消費(fèi)完成時(shí)，數(shù)據(jù)落地到 Hive 中可能是小時(shí)級(jí)的或者是半小時(shí)級(jí)的，甚至是分鐘級(jí)的，此時(shí)需要知道數(shù)據(jù)的 Event time 已經(jīng)到了什么時(shí)間，然后再去觸發(fā)比如 alert table、add partition、 add location 等，把分區(qū)寫(xiě)進(jìn) Hive 中。這時(shí)還需要看一下當(dāng)前的 Flink 任務(wù)的數(shù)據(jù)時(shí)間消費(fèi)到了什么時(shí)間，如9點(diǎn)的數(shù)據(jù)要落地時(shí)，需要看一下 Kafka 里 Flink 數(shù)據(jù)消費(fèi)是否到了9點(diǎn)，然后在 Hive 中觸發(fā)分區(qū)寫(xiě)入。

2. 實(shí)現(xiàn)原理

這塊的實(shí)現(xiàn)原理主要是使用 Flink 高階版本的特性 StreamingFileSink。

StreamingFileSink 的主要功能如下：

• forBulkFormat 支持 avro、parquet 格式，也就是支持鏈?zhǔn)降拇鎯?chǔ)格式

• withBucketAssigner 自定義按數(shù)據(jù)時(shí)間分桶，支持?jǐn)?shù)據(jù)時(shí)間的分桶，上圖用到該功能的地方定義了一個(gè) EventtimeBucket，按照數(shù)據(jù)的時(shí)間落地到離線中

• OnCheckpointRollingPolicy，會(huì)根據(jù) CheckPoint 時(shí)間來(lái)進(jìn)行數(shù)據(jù)的落地，此處可以理解為按照數(shù)據(jù)的時(shí)間，比如按照一定的 CheckPoint 時(shí)間內(nèi)進(jìn)行數(shù)據(jù)落地、回滾，數(shù)據(jù)落地策略還可以按照數(shù)據(jù)大小落地

• Exactly-Once 語(yǔ)義實(shí)現(xiàn)，F(xiàn)link 中自帶的 StreamingFileSink 是用 Exactly-Once 語(yǔ)義來(lái)實(shí)現(xiàn)的。Flink 中有兩個(gè) Exactly-Once 的實(shí)現(xiàn)，第一個(gè)是 Kafka 的 Exactly-Once，第二個(gè)是 StreamingFileSink 實(shí)現(xiàn)了 Exactly-Once 語(yǔ)義，像上圖中 CheckpointRollingPolicy 設(shè)置的是十分鐘落地一次到 HDFS 文件中

下面來(lái)具體說(shuō)一下 Exactly-Once 是如何實(shí)現(xiàn)的。

① Exactly-Once

具體實(shí)現(xiàn) Exactly-Once 的方式，如上圖所示，左側(cè)是一個(gè)二階段的模型，Coordinator 發(fā)一個(gè) perpare，所有的參與者或者執(zhí)行者開(kāi)始觸發(fā) ack 動(dòng)作，Coordinator 收到所有人的 ack 動(dòng)作后，就開(kāi)始執(zhí)行 commit，所有的執(zhí)行者就把左右的數(shù)據(jù)進(jìn)行落地。到了 Flink 這塊，Source 收到了 checkpoint barrier 流的時(shí)候，開(kāi)始觸發(fā) snapshorState 發(fā)送到 Job Manager，Job Manager 把所有的 CheckPoint 都完成以后，會(huì)發(fā)送一個(gè) notifyCheckpointComplete，F(xiàn)link 這塊跟上圖左邊的二階段提交協(xié)議是一致的，F(xiàn)link 也是可以實(shí)現(xiàn)二階段提交協(xié)議的。

② 如何使用 Flink 實(shí)現(xiàn)二階段提交協(xié)議

首先 StramingFileSink 實(shí)現(xiàn)了兩個(gè)接口，分別是 CheckpointedFunction 和 CheckpointListener。

• CheckpointedFunction 實(shí)現(xiàn)了 initialzeState 和 snaoshotState 這兩個(gè)函數(shù)；

• CheckpointListener 是 notifyCheckPoint Complete 的方法實(shí)現(xiàn)。

所以這兩個(gè)接口可以實(shí)現(xiàn)二階段提交的語(yǔ)義，initialzeState 算子剛啟動(dòng)的時(shí)候，它會(huì)啟動(dòng)三個(gè)動(dòng)作 commitpendingFile、restoreInProgressFile、truncate。

第一步 commitpedingFile，也就是實(shí)時(shí)的數(shù)據(jù)落地到 HDFS 的時(shí)候，有三個(gè)狀態(tài)，第一個(gè)狀態(tài)是 in-progress，即正在進(jìn)行中的一個(gè)狀態(tài)，第二個(gè)狀態(tài)是 pending 的狀態(tài)，第三個(gè)狀態(tài)是 finish 的狀態(tài)。

在實(shí)時(shí)的寫(xiě)入時(shí)，如果 CheckPoint 還沒(méi)有在這之間成功的時(shí)候，程序出問(wèn)題了，那接下來(lái)啟動(dòng)的時(shí)候就會(huì)觸發(fā) initialzeState，會(huì)把曾經(jīng) pending 的 file 進(jìn)行 commit，然后把寫(xiě)了一半的文件比如 in-progress 文件重置或者截?cái)啵M(jìn)行重置或者截?cái)嗍鞘褂玫氖?Hadoop 的2.7版本的 turncate 方式。也就是數(shù)據(jù)在一直寫(xiě)入，但是寫(xiě)入沒(méi)有達(dá)到一個(gè) CheckPoint 周期，也就是說(shuō)中間數(shù)據(jù)斷開(kāi)了，下一次啟動(dòng)的時(shí)候，要么把之前沒(méi)有寫(xiě)完整的數(shù)據(jù)截?cái)嗟?，之?CheckPoint 觸發(fā)已經(jīng)寫(xiě)好的數(shù)據(jù)直接 commit。

第二步 invoke 就是數(shù)據(jù)實(shí)時(shí)的寫(xiě)入

第三步 snapshotState 在觸發(fā) CheckPoint 的時(shí)候會(huì)把 in-progress 文件轉(zhuǎn)成 pending state 文件，也就是開(kāi)始提交文件，同時(shí)記錄 length 長(zhǎng)度。記錄長(zhǎng)度是因?yàn)榍斑叺牟襟E需要 truncate 來(lái)截?cái)喽嚅L(zhǎng)，snapshot 時(shí)，是沒(méi)有真正的寫(xiě)入到 HDFS，其實(shí)是寫(xiě)入到 ListState，等所有的 CheckPoint 算子都完成了，就把 ListState 中的數(shù)據(jù)都刷到 HDFS 中，只要數(shù)據(jù)存在 Flink 自帶的 state 中，不斷把數(shù)據(jù)成功的刷到 HDFS 中就行了。

第四步 notifyCheckPoint Complete 會(huì)觸發(fā) pending 動(dòng)作到 finished 狀態(tài)的數(shù)據(jù)寫(xiě)入，實(shí)現(xiàn)的方式直接使用 rename，Streaming 會(huì)不斷的寫(xiě)入 HDFS 中的臨時(shí)文件，等到 notifyCheckPoint 結(jié)束之后，直接做一個(gè) rename 動(dòng)作，寫(xiě)成正式文件。

3. 跨集群多 nameservices

趣頭條的實(shí)時(shí)集群跟離線集群是獨(dú)立的，實(shí)時(shí)集群目前是一套，離線集群是有多套。通過(guò)實(shí)時(shí)集群要寫(xiě)入到離線集群，這樣就會(huì)遇到一個(gè)問(wèn)題，HDFS nameservices 問(wèn)題，如果在實(shí)時(shí)集群中把所有的離線集群的 nameservice 用 namenode HA 的方式全部打入到實(shí)時(shí)集群，是不太合適的。所以使用 Flink 任務(wù)中 resource 下邊把 HDFS 中的 xml 文件中間加 final 標(biāo)簽，設(shè)置為 true。此處的 value 標(biāo)簽中，stream 是一個(gè)實(shí)時(shí)集群，date 是一個(gè)離線集群，這樣把兩個(gè) HA 配置在 value 標(biāo)簽，從而達(dá)到實(shí)時(shí)集群是實(shí)時(shí)集群，離線集群是離線集群，中間的 HDFS 中 set 不需要相互修改，直接在客戶端時(shí)間就行了。

4. 多用戶寫(xiě)入權(quán)限

針對(duì)多用戶權(quán)限寫(xiě)入的問(wèn)題，實(shí)時(shí)寫(xiě)入離線 HDFS 中的時(shí)候，會(huì)涉及到用戶權(quán)限。遇到用戶權(quán)限時(shí)，也會(huì)有一個(gè)問(wèn)題，F(xiàn)link 實(shí)時(shí)提交的用戶，是定義好的，所有的程序里用戶是同一個(gè)，但是離線是多個(gè)用戶，F(xiàn)link 目前對(duì)于這塊用戶的權(quán)限做的還不夠好，所以我們自己改造了一下，在 API 中添加了 withBucketUser，上邊已經(jīng)配置好了 nameServices，然后通過(guò)該參數(shù)來(lái)配置具體是那個(gè)用戶來(lái)寫(xiě)入 HDFS 中，這是 API 層級(jí)的。

API 層級(jí)的好處是一個(gè) Flink 程序可以寫(xiě)多個(gè)，可以指定不同的 HDFS 的不同的用戶就可以。具體實(shí)現(xiàn)就是在 Hadoop file system 中加一個(gè) ugi.do as，代理用戶。以上是趣頭條用 Flink 在實(shí)時(shí)數(shù)據(jù)同步到 Hive 做的一些工作。其中會(huì)有一些小文件的問(wèn)題，針對(duì)小文件，我們通過(guò)后臺(tái)程序定期的 merge，如果 CheckPoint 的時(shí)間很短，就會(huì)出現(xiàn)大量的小文件的問(wèn)題。

1. 秒級(jí)實(shí)現(xiàn)架構(gòu)圖

首先來(lái)解釋一下趣頭條使用 Flink+ClickHouse 的場(chǎng)景，最開(kāi)始展示的很多實(shí)時(shí)指標(biāo)，可能是每五分鐘計(jì)算一次，也可能是每三分鐘計(jì)算一次。如果每一個(gè)實(shí)時(shí)指標(biāo)用一個(gè) Flink 任務(wù)，即使是 FlinkSQL 來(lái)寫(xiě)，比如消費(fèi)一個(gè) Kafka Topic，計(jì)算它的日活、新增、流程等，當(dāng)用戶提出一個(gè)新的需求，那這個(gè) Flink 任務(wù)是需要修改還是再啟動(dòng)一個(gè) Flink 任務(wù)來(lái)消費(fèi)這個(gè) Topic，這樣的話就會(huì)出現(xiàn) Flink 任務(wù)在不斷的修改或者不斷的啟動(dòng)新的 Flink 新的任務(wù)。為了解決這個(gè)問(wèn)題，就讓 Flink 后邊接一個(gè)套 ClickHouse 實(shí)現(xiàn)整體的 OLAP。

上圖為秒級(jí)實(shí)現(xiàn)架構(gòu)圖，從 Kafka 到 Flink 到 Hive 然后再到 ClickHouse 集群，對(duì)接外部 Horizon ( 實(shí)時(shí)報(bào)表 )、QE ( 實(shí)時(shí) adhoc 查詢 )、千尋 ( 數(shù)據(jù)分析 )、用戶畫(huà)像 ( 實(shí)時(shí)的用戶畫(huà)像 )。

2. Why Flink+ClickHouse

具體來(lái)說(shuō)為什么要用 Flink+ClickHouse，主要有以下幾點(diǎn)：

• 指標(biāo)實(shí)現(xiàn)支持 sql 描述，以前的方案使用是 storm 的程序，通過(guò) stormsql 實(shí)現(xiàn)，包括 flinksql，這些內(nèi)容對(duì)于 UDF 支持相對(duì)有限，但是現(xiàn)在這套 Flink+ClickHouse 基本上可以把分析師提的指標(biāo)通過(guò) sql 實(shí)現(xiàn)。

• 指標(biāo)的上下線互不影響，這個(gè)主要是解決上邊提到的關(guān)于 Flink 任務(wù)消費(fèi)了 topic 以后，假如用戶提出新的指標(biāo)的時(shí)候，是啟動(dòng)新任務(wù)還是要不斷修改的問(wèn)題。

• 數(shù)據(jù)可回溯，方便異常排查，這個(gè)就類似上邊提到的假如我的日活掉了，需要知道哪些指標(biāo)的口徑的邏輯掉了、哪個(gè)上報(bào)的數(shù)據(jù)掉了，如 cmd 掉了還是數(shù)據(jù)流 kafka 掉了還是用戶上報(bào)的時(shí)候指標(biāo)沒(méi)有上報(bào)導(dǎo)致的日活掉了。假如單純的 flink 的話，只是會(huì)計(jì)算出那個(gè)指標(biāo)掉了，是沒(méi)辦法回溯的。

• 計(jì)算快，一個(gè)周期內(nèi)完成所有的指標(biāo)計(jì)算，現(xiàn)在的 horizon 曲線可能是幾百上千，需要在五分鐘之內(nèi)或者十分鐘之內(nèi)，把所有分時(shí)、累時(shí)、以及維度下降的指標(biāo)全部計(jì)算出來(lái)。

• 支持實(shí)時(shí)流，分部署部署，運(yùn)維簡(jiǎn)單。

目前趣頭條 Flink 集群有 100+ 臺(tái) 32 核 128 G 3.5T SSD，日數(shù)據(jù)量 2000+ 億，日查詢量 21w+ 次，80% 查詢?cè)?1s 內(nèi)完成。

上圖為單表測(cè)試結(jié)果。ClickHouse 單表測(cè)試速度快。但受制于架構(gòu)，ClickHouse 的 Join 較弱。

上圖是處理相對(duì)較為復(fù)雜的 SQL，count+group by+order by，ClickHouse 在 3.6s內(nèi)完成 26 億數(shù)據(jù)計(jì)算。

3. Why ClickHouse so Fast

接下來(lái)說(shuō)一下為什么 ClickHouse 這么快，主要是有以下幾點(diǎn)：

• 列式存儲(chǔ)+LZ4、ZSTD 數(shù)據(jù)壓縮：列式存儲(chǔ)基本是通用的。

• 計(jì)算存儲(chǔ)本地化+向量化執(zhí)行：計(jì)算存儲(chǔ)本地化，ClickHouse 跟 presto 不一樣，presto 數(shù)據(jù)可能存在 Hadoop 集群里邊或者 HDFS 中，需要把數(shù)據(jù)拉過(guò)來(lái)，然后進(jìn)行實(shí)時(shí)的計(jì)算；而 ClickHouse 是每一臺(tái)計(jì)算機(jī)器需要的數(shù)據(jù)存儲(chǔ)在本地的 ssd 盤(pán)，只要計(jì)算本地的數(shù)據(jù)就可以了，比如求 count 之類的，計(jì)算完成后把其他的節(jié)點(diǎn)進(jìn)行合并就可以了。

• LSM merge tree+Index：LSM merge tree，他會(huì)不斷的使用 batch 的形式把數(shù)據(jù)寫(xiě)入到 ClickHouse 之后，在后臺(tái)做了一個(gè)線程把數(shù)據(jù)進(jìn)行 merge，做一個(gè) index 索引，也就是給這張數(shù)據(jù)表建立很多索引，類如常見(jiàn)的 DT 的時(shí)間索引、小時(shí)級(jí)的數(shù)據(jù)索引來(lái)提高查詢性能或者速度。

• SIMD+LLVM 優(yōu)化：SIMD 就是一個(gè)單指令多數(shù)據(jù)集，LLVM 是一個(gè) C++ 的編譯器

• SQL 語(yǔ)法、UDF 完善：在這塊有很大的需求，比如數(shù)據(jù)分析以及維度下墜，常規(guī)的 horizon 數(shù)據(jù)報(bào)表可能就是 count、sum、以及 group by、order by 等，但是在一些維度下墜或者是數(shù)據(jù)分析領(lǐng)域，可能會(huì)有一個(gè)窗口期的概念，在一段窗口期內(nèi)的留存，所以要用到一些更高的特性，類如時(shí)間窗口的功能。

上圖是 MergeTree 的運(yùn)行原理圖解，最上邊的第一層是數(shù)據(jù)一個(gè) batch 一個(gè) batch 的實(shí)時(shí)寫(xiě)入，后臺(tái)會(huì)做每一個(gè)層級(jí)的數(shù)據(jù) merge，這塊跟 HBase 差不多的實(shí)現(xiàn)，merge 的時(shí)候會(huì)進(jìn)行數(shù)據(jù)的排序，然后做一個(gè)數(shù)據(jù)索引。

上圖是 ClickHouse Connector，ClickHouse 有兩個(gè)概念，local table 和 distribute table。local table 是用來(lái)寫(xiě)的，當(dāng)然 distribute table 也可以寫(xiě)入，但是會(huì)出現(xiàn)很大的 io 問(wèn)題，所以盡量不要寫(xiě) distribute table。但是可以讀 distribute table。5-10w 一個(gè) batch 進(jìn)行數(shù)據(jù)寫(xiě)入，正常的情況下，是5秒一個(gè)周期。

RoundRobinClickHouse DataSource 這塊是趣頭條自己實(shí)現(xiàn)的；

ClickHouse 官方 API 使用：

BalancedClickHouseDataSource 實(shí)現(xiàn)的。

上圖是 ClickHouse 官方 API 使用：

BalancedClickHouseDataSource

里邊有一個(gè)問(wèn)題，比如 mysql 配置一個(gè) ip 和端口號(hào)就可以把數(shù)據(jù)寫(xiě)入了，但是這塊要寫(xiě)入 local table 的，所以必須要知道這一個(gè)集群到底有多少 local table，每一個(gè) local table 的 ip 和端口號(hào)，假如有100臺(tái)機(jī)器，就必須要把這100臺(tái)機(jī)器的 ip 和端口號(hào)配置好，然后進(jìn)行寫(xiě)入。

官方的 api 中有兩個(gè) schedule：

• 一個(gè)是 scheduleActualization

• 另一個(gè)是 scheduleConnectionsCleaning

第一個(gè)是指100臺(tái)機(jī)器配置了100個(gè) ip 或者端口號(hào)，可能會(huì)有一些機(jī)器出現(xiàn) ping 不通或者服務(wù)無(wú)響應(yīng)，這塊是定時(shí)的做一個(gè) Actualiza 來(lái)發(fā)現(xiàn)這些機(jī)器哪些無(wú)法連接，觸發(fā)一個(gè)下限來(lái)把這些 ip 刪除掉。

第二個(gè) scheduleConnectionsCleaning，因?yàn)?ClickHouse 是 http 的方式，定期的會(huì)把一些沒(méi)用的 http 的請(qǐng)求清理掉。

針對(duì)于官方提供的 API，趣頭條對(duì)這方面做了一個(gè)加強(qiáng)，開(kāi)發(fā)了一個(gè) RoundRobinClickHouseDataSource，實(shí)現(xiàn)了三個(gè)語(yǔ)義，分別是 testOnBorrow、testOnReturn、testWhileldle。

第一個(gè) testOnBorrow 取鏈接的時(shí)候，設(shè)置 為true，然后去 ping 一下這個(gè)鏈接能不能拿到，ClickHouse 寫(xiě)入的時(shí)候，使用的 batch，所以盡量就是拿鏈接的時(shí)候要拿到成功的鏈接；第二個(gè) testOnReturn 設(shè)置為 false，testWhileldle 設(shè)置為 true，把上邊官方的兩個(gè) schedule 功能集成進(jìn)去了。為什么要實(shí)現(xiàn) RoundRobin，主要是因?yàn)榧偃缬?00臺(tái)機(jī)器，ClickHouse 相對(duì)于 Hadoop 來(lái)說(shuō)，還是需要好好維護(hù)一下，如果是 insert 的話，后臺(tái)是不斷 merge 的過(guò)程，insert 速度大于 merge 速度時(shí)候，會(huì)導(dǎo)致 merge 速度永遠(yuǎn)跟不上，所以就寫(xiě)完這臺(tái)機(jī)器接下來(lái)寫(xiě)別的機(jī)器，以及5秒一個(gè)間隔的寫(xiě)，使 merge 的速度盡量跟上 insert 的速度，這塊是整個(gè)部分最需要注意的地方。

4. Backfill

趣頭條針對(duì)集群容錯(cuò)做了一些優(yōu)化，主要包括兩點(diǎn)：

• 第一點(diǎn)是 Flink 任務(wù)小時(shí)級(jí)容錯(cuò)

• 第二點(diǎn)是 ClickHouse 集群小時(shí)級(jí)容錯(cuò)

Flink 導(dǎo)入數(shù)據(jù)到 ClickHouse，來(lái)實(shí)現(xiàn)數(shù)據(jù)的查詢、報(bào)表展示，會(huì)遇到一些問(wèn)題。如 Flink 任務(wù)出現(xiàn)故障、報(bào)錯(cuò)、數(shù)據(jù)反壓、network 的一些問(wèn)題；或者 ClickHouse 集群出現(xiàn)了一些不可響應(yīng)、ZK 跟不上等 ZK 問(wèn)題；或者集群的負(fù)載問(wèn)題；或者是上邊提到的 insert 太快的問(wèn)題；會(huì)導(dǎo)致整個(gè)任務(wù)都有問(wèn)題。如果數(shù)據(jù)量突然暴漲，把 Flink 啟動(dòng)，就會(huì)出現(xiàn)一段時(shí)間內(nèi)不停的追數(shù)據(jù)，可能就需要調(diào)大它的并行度之類的，讓 Flink 任務(wù)把數(shù)據(jù)追上。但是數(shù)據(jù)已經(jīng)積壓了，F(xiàn)link 又要加大它的并發(fā)度來(lái)處理數(shù)據(jù)，但是 ClickHouse 那塊又限制了 insert 速度不能太快，所以就做了另外一個(gè)機(jī)制，也就是 Flink 故障了或者 ClickHouse 故障了，等到 ClickHouse 集群恢復(fù)之后，F(xiàn)link 任務(wù)還是從最新的開(kāi)始消費(fèi)，過(guò)去的一段數(shù)據(jù)不再去追了，通過(guò) Hive 來(lái)把數(shù)據(jù)導(dǎo)入到 ClickHouse。

用 Hive 是因?yàn)閿?shù)據(jù)通過(guò) Kafka 已經(jīng)實(shí)時(shí)落地到 Hive，通過(guò) waterdrop 把數(shù)據(jù)寫(xiě)入到 ClickHouse，ClickHouse 是有分區(qū)的，只要把上一個(gè)小時(shí)的數(shù)據(jù)刪除，再把 Hive 一個(gè)小時(shí)的數(shù)據(jù)導(dǎo)入進(jìn)來(lái)，這樣就可以繼續(xù)提供數(shù)據(jù)查詢操作了。

最后是對(duì)未來(lái)的發(fā)展與思考。

1. Connector SQL

對(duì)于未來(lái)的發(fā)展，首先是 Connectors SQL，也就是把 Connector 進(jìn)行 SQL 化，現(xiàn)在是 Flink-to-Hive 以及 Flink-to-ClickHouse，相對(duì)來(lái)講，都是比較固化的一些場(chǎng)景，所以是可以進(jìn)行 sql 化，除了把 HDFS 的路徑指定以及用戶指定，其他的一些過(guò)程都是可以 SQL 化描述出來(lái)的。

2. Delta lake

Flink 是流批一體計(jì)算引擎，但是沒(méi)有流批一體的存儲(chǔ)。趣頭條會(huì)用 HBase、Kudu、Redis 等能夠與 Flink 實(shí)時(shí)交互的 KV 存儲(chǔ)進(jìn)行數(shù)據(jù)計(jì)算。如計(jì)算新增問(wèn)題，目前趣頭條的方案是需要將 Hive 歷史用戶刷到 Redis 或 HBase 中，與 Flink 進(jìn)行實(shí)時(shí)交互判斷用戶是否新增。但因?yàn)?Hive 中的數(shù)據(jù)和 Redis 中的數(shù)據(jù)是存儲(chǔ)為兩份數(shù)據(jù)。其次 Binlog 抽取數(shù)據(jù)會(huì)涉及 delete 動(dòng)作，Hbase，Kudu 支持?jǐn)?shù)據(jù)修改，定期回到 Hive 中。帶來(lái)的問(wèn)題是 HBase，Kudu 中存在數(shù)據(jù)，Hive 又保存了一份數(shù)據(jù)，多出一份或多份數(shù)據(jù)。如果有流批一體的存儲(chǔ)支持上述場(chǎng)景，當(dāng) Flink 任務(wù)過(guò)來(lái)，可以與離線數(shù)據(jù)進(jìn)行實(shí)時(shí)交互，包括實(shí)時(shí)查詢 Hive 數(shù)據(jù)等，可以實(shí)時(shí)判斷用戶是否新增，對(duì)數(shù)據(jù)進(jìn)行實(shí)時(shí)修改、更新或 delete，也能支持 Hive 的批的動(dòng)作存儲(chǔ)。未來(lái)，趣頭條考慮對(duì) Flink 做流批的存儲(chǔ)，使 Flink 生態(tài)統(tǒng)一為流批結(jié)合。

嘉賓介紹：

王金海，10 年互聯(lián)網(wǎng)歷練，先后在唯品會(huì)負(fù)責(zé)用戶畫(huà)像系統(tǒng)，提供人群的個(gè)性化營(yíng)銷服務(wù)；餓了么擔(dān)任架構(gòu)師，負(fù)責(zé)大數(shù)據(jù)任務(wù)調(diào)度、元數(shù)據(jù)開(kāi)發(fā)、任務(wù)畫(huà)像等工作；現(xiàn)為趣頭條數(shù)據(jù)中心平臺(tái)負(fù)責(zé)人，負(fù)責(zé)大數(shù)據(jù)基礎(chǔ)計(jì)算層 ( spark、presto、flink、clickhouse )、平臺(tái)服務(wù)層 ( libra 實(shí)時(shí)計(jì)算、kepler 離線調(diào)度 )、數(shù)據(jù)產(chǎn)品層 ( qe即時(shí)查詢、horizon 數(shù)據(jù)報(bào)表、metadata 元數(shù)據(jù)、數(shù)據(jù)權(quán)限等 )、以及團(tuán)隊(duì)建設(shè)。

今天的分享就到這里，謝謝大家。

特別推薦一個(gè)分享架構(gòu)+算法的優(yōu)質(zhì)內(nèi)容，還沒(méi)關(guān)注的小伙伴，可以長(zhǎng)按關(guān)注一下：
長(zhǎng)按訂閱更多精彩▼
serif;letter-spacing: 0.544px;white-space: normal;text-align: right;line-height: 2em;box-sizing: border-box !important;word-wrap: break-word !important;">如有收獲，點(diǎn)個(gè)在看，誠(chéng)摯感謝