如何基于Flink+ClickHouse 構建實時數據分析平臺

今天就跟大家聊聊有關如何基于Flink+ClickHouse 構建實時數據分析平臺，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結了以下內容，希望大家根據這篇文章可以有所收獲。

• 第一，  forBulkFormat 支持 avro、parquet 格式，即列式存儲格式。

• 第二，  withBucketAssigner 自定義按數據時間分桶，此處會定義一個EventtimeBucket，既按數據時間進行數據落地到離線中。

• 第三，  OnCheckPointRollingPolicy，根據 CheckPoint 時間進行數據落地，在一定的 CheckPoint 時間內數據落地并回穩。按照 CheckPoint 落地還有其它策略，如按照數據大小。

• 第四，  StreamingFileSink 是 Exactly-Once 語義實現。

• initializeState 

• invoke 

• snapshotState 

• notifyCheckpointComplete 

• 指標實現 sql 化描述：分析師提出的指標基本都以 SQL 進行描述。

• 指標的上下線互不影響：一個 Flink 任務消費 Topic，如果還需要其它指標，可以保證指標的上下線互不影響。

• 數據可回溯，方便異常排查：當日活下降，需要回溯排查是哪些指標口徑的邏輯問題，比如是報的數據差異或是數據流 Kafka 掉了，或者是因為用戶沒有上報某個指標導致日活下降，而 Flink 則無法進行回溯。

• 計算快，一個周期內完成所有指標計算：需要在五分鐘內將成百上千的所有維度的指標全部計算完成。

• 支持實時流，分布式部署，運維簡單：支持 Kafka 數據實時流。

• Merge Tree：如下圖所示。第一層為實時數據寫入。后臺進行每一層級數據的merge。merge 時會進行數據排序，做 Index 索引。

• ClickHouse Connector：ClickHouse 有兩個概念，Local table 和Distributed table。一般是寫 Local table ，讀 Distributed table。ClickHouse 一般以 5~10w一個批次進行數據寫入，5s一個周期。趣頭條還實現了 RoundRobinClickHouseDataSource。

• BalancedClickHouseDataSource ：MySQL 中配置一個 IP 和端口號就可以寫入數據，而 BalancedClickHouseDataSource 需要寫 Local 表，因此必須要知道該集群有多少個 Local 表，每一個 Local 表的 IP 和端口號。如有一百臺機器，需要將一百臺機器的 IP 和端口號全部配置好，再進行寫入。BalancedClickHouseDataSource 有兩個 schedule。scheduleActualization和 scheduleConnectionsCleaning 。配置一百臺機器的 IP 和端口號，會出現某些機器不連接或者服務不響應問題，scheduleActualization 會定期發現機器無法連接的問題，觸發下線或刪除 IP 等動作。scheduleConnectionsCleaning 會定期清理 ClickHouse 中無用的 http 請求。

• RoundRobinClickHouseDataSource：趣頭條對BalancedClickHouseDataSource 進行加強的結果，實現了三個語義。testOnBorrow 設置為 true，嘗試 ping 看能否獲取連接。用 ClickHouse 寫入時是一個 batch，再將 testOnReturn 設置為 false，testWhileIdel 設置為true，填入官方 scheduleActualization 和 scheduleConnectionsCleaning 的功能。ClickHouse 后臺不斷進行 merge，如果 insert 過快使后臺 merge 速度變慢，跟不上 insert，出現報錯。因此需要盡量不斷往下寫，等寫完當前機器，再寫下一個機器，以5s間隔進行寫入，使 merge 速度能夠盡量與 insert 速度保持一致。

看完上述內容，你們對如何基于Flink+ClickHouse 構建實時數據分析平臺有進一步的了解嗎？如果還想了解更多知識或者相關內容，請關注億速云行業資訊頻道，感謝大家的支持。