簡介：本文由網(wǎng)易云音樂實時計算平臺研發(fā)工程師岳猛分享，主要從以下四個部分將為大家介紹 Flink + Kafka 在網(wǎng)易云音樂的應用實戰(zhàn)：

背景

Flink + Kafka 平臺化設(shè)計

Kafka 在實時數(shù)倉中的應用

問題 & 改進

直播回放：https://developer.aliyun.com/live/2894

一、背景介紹

(一)流平臺通用框架目前流平臺通用的架構(gòu)一般來說包括消息隊列、計算引擎和存儲三部分，通用架構(gòu)如下圖所示。客戶端或者 web 的 log 日志會被采集到消息隊列；計算引擎實時計算消息隊列的數(shù)據(jù)；實時計算結(jié)果以 Append 或者 Update 的形式存放到實時存儲系統(tǒng)中去。目前，我們常用的消息隊列是 Kafka，計算引擎一開始我們采用的是 Spark Streaming，隨著 Flink 在流計算引擎的優(yōu)勢越來越明顯，我們最終確定了 Flink 作為我們統(tǒng)一的實時計算引擎。

(二)為什么選 Kafka？Kafka 是一個比較早的消息隊列，但是它是一個非常穩(wěn)定的消息隊列，有著眾多的用戶群體，網(wǎng)易也是其中之一。我們考慮 Kafka 作為我們消息中間件的主要原因如下：

• 高吞吐，低延遲：每秒幾十萬 QPS 且毫秒級延遲；

• 高并發(fā)：支持數(shù)千客戶端同時讀寫；

• 容錯性，可高性：支持數(shù)據(jù)備份，允許節(jié)點丟失；

• 可擴展性：支持熱擴展，不會影響當前線上業(yè)務。

(三)為什么選擇 Flink？Apache Flink 是近年來越來越流行的一款開源大數(shù)據(jù)流式計算引擎，它同時支持了批處理和流處理，考慮 Flink 作為我們流式計算引擎的主要因素是：

• 高吞吐，低延遲，高性能；

• 高度靈活的流式窗口；

• 狀態(tài)計算的 Exactly-once 語義；

• 輕量級的容錯機制；

• 支持 EventTime 及亂序事件；

• 流批統(tǒng)一引擎。

(四)Kafka + Flink 流計算體系基于 Kafka 和 Flink 的在消息中間件以及流式計算方面的耀眼表現(xiàn)，于是產(chǎn)生了圍繞 Kafka 及 Flink 為基礎(chǔ)的流計算平臺體系，如下圖所示：基于 APP、web 等方式將實時產(chǎn)生的日志采集到 Kafka，然后交由 Flink 來進行常見的 ETL，全局聚合以及Window 聚合等實時計算。

(五)網(wǎng)易云音樂使用 Kafka 的現(xiàn)狀目前我們有 10+個 Kafka 集群，各個集群的主要任務不同，有些作為業(yè)務集群，有些作為鏡像集群，有些作為計算集群等。當前 Kafka 集群的總節(jié)點數(shù)達到 200+，單 Kafka 峰值 QPS 400W+。目前，網(wǎng)易云音樂基于 Kafka+Flink 的實時任務達到了 500+。

二、Flink+Kafka 平臺化設(shè)計

基于以上情況，我們想要對 Kafka+Flink 做一個平臺化的開發(fā)，減少用戶的開發(fā)成本和運維成本。實際上在 2018 年的時候我們就開始基于 Flink 做一個實時計算平臺，Kafka 在其中發(fā)揮著重要作用，今年，為了讓用戶更加方便、更加容易的去使用 Flink 和 Kafka，我們進行了重構(gòu)。基于 Flink 1.0 版本我們做了一個 Magina 版本的重構(gòu)，在 API 層次我們提供了 Magina SQL 和 Magina SDK 貫穿 DataStream 和 SQL 操作；然后通過自定義 Magina SQL Parser 會把這些 SQL 轉(zhuǎn)換成 Logical Plan，在將 LogicalPlan 轉(zhuǎn)化為物理執(zhí)行代碼，在這過程中會去通過 catalog 連接元數(shù)據(jù)管理中心去獲取一些元數(shù)據(jù)的信息。我們在 Kafka 的使用過程中，會將 Kafka 元數(shù)據(jù)信息登記到元數(shù)據(jù)中心，對實時數(shù)據(jù)的訪問都是以流表的形式。在 Magina 中我們對 Kafka 的使用主要做了三部分的工作：

• 集群 catalog 化；

• Topic 流表化；

• Message Schema 化。

用戶可以在元數(shù)據(jù)管理中心登記不同的表信息或者 catalog 信息等，也可以在 DB 中創(chuàng)建和維護 Kafka 的表，用戶在使用的過程只需要根據(jù)個人需求使用相應的表即可。下圖是對 Kafka 流表的主要引用邏輯。

三、Kafka 在實時數(shù)倉中的應用

(一)在解決問題中發(fā)展Kafka 在實時數(shù)倉使用的過程中，我們遇到了不同的問題，中間也嘗試了不同的解決辦法。在平臺初期， 最開始用于實時計算的只有兩個集群，且有一個采集集群，單 Topic 數(shù)據(jù)量非常大；不同的實時任務都會消費同一個大數(shù)據(jù)量的 Topic，Kafka 集群 IO 壓力異常大；因此，在使用的過程發(fā)現(xiàn) Kafka 的壓力異常大，經(jīng)常出現(xiàn)延遲、I/O 飆升。我們想到把大的 Topic 進行實時分發(fā)來解決上面的問題，基于 Flink 1.5 設(shè)計了如下圖所示的數(shù)據(jù)分發(fā)的程序，也就是實時數(shù)倉的雛形。基于這種將大的 Topic 分發(fā)成小的 Topic 的方法，大大減輕了集群的壓力，提升了性能，另外，最初使用的是靜態(tài)的分發(fā)規(guī)則，后期需要添加規(guī)則的時候要進行任務的重啟，對業(yè)務影響比較大，之后我們考慮了使用動態(tài)規(guī)則來完成數(shù)據(jù)分發(fā)的任務。

解決了平臺初期遇到的問題之后，在平臺進階過程中 Kafka 又面臨新的問題：

• 雖然進行了集群的擴展，但是任務量也在增加，Kafka 集群壓力仍然不斷上升；

• 集群壓力上升有時候出現(xiàn) I/O 相關(guān)問題，消費任務之間容易相互影響；

• 用戶消費不同的 Topic 過程沒有中間數(shù)據(jù)的落地，容易造成重復消費；

• 任務遷移 Kafka 困難。

針對以上問題，我們進行了如下圖所示的 Kafka 集群隔離和數(shù)據(jù)分層處理。其過程簡單來說，將集群分成 DS 集群、日志采集集群、分發(fā)集群，數(shù)據(jù)通過分發(fā)服務分發(fā)到 Flink 進行處理，然后通過數(shù)據(jù)清洗進入到 DW 集群，同時在 DW 寫的過程中會同步到鏡像集群，在這個過程中也會利用 Flink 進行實時計算的統(tǒng)計和拼接，并將生成的 ADS 數(shù)據(jù)寫入在線 ADS 集群和統(tǒng)計 ADS 集群。通過上面的過程，確保了對實時計算要求比較高的任務不會受到統(tǒng)計報表的影響。

通過上面的過程，確保了對實時計算要求比較高的任務不會受到統(tǒng)計報表的影響。但是我們分發(fā)了不同的集群以后就不可避免的面臨新的問題：

• 如何感知 Kafka 集群狀態(tài)？

• 如何快速分析 Job 消費異常？

針對上面兩個問題，我們做了一個 Kafka 監(jiān)控系統(tǒng)，其監(jiān)控分為如下兩個維度，這樣在出現(xiàn)異常的時候就可以進行具體判斷出現(xiàn)問題的詳細情況：

• 集群概況的監(jiān)控：可以看到不同集群對應的 Topic 數(shù)量以及運行任務數(shù)量，以及每個 Topic 消費任務數(shù)據(jù)量、數(shù)據(jù)流入量、流入總量和平均每條數(shù)據(jù)大小；

• 指標監(jiān)控：可以看到 Flink 任務以及對應的 Topic、GroupID、所屬集群、啟動時間、輸入帶寬、InTPS、OutTPS、消費延遲以及 Lag 情況。

(二)Flink + Kafka 在 Lambda 架構(gòu)下的運用流批統(tǒng)一是目前非常火的概念，很多公司也在考慮這方面的應用，目前常用的架構(gòu)要么是 Lambda 架構(gòu)，要么是 Kappa 架構(gòu)。對于流批統(tǒng)一來講需要考慮的包括存儲統(tǒng)一和計算引擎統(tǒng)一，由于我們當前基建沒有統(tǒng)一的存儲，那么我們只能選擇了 Lamda 架構(gòu)。下圖是基于 Flink 和 Kafka 的 Lambda 架構(gòu)在云音樂的具體實踐，上層是實時計算，下層是離線計算，橫向是按計算引擎來分，縱向是按實時數(shù)倉來區(qū)分。

四、問題&改進

在具體的應用過程中，我們也遇到了很多問題，最主要的兩個問題是：

• 多 Sink 下 Kafka Source 重復消費問題；

• 同交換機流量激增消費計算延遲問題。

(一)多 Sink 下 Kafka Source 重復消費問題Magina 平臺上支持多 Sink，也就是說在操作的過程中可以將中間的任意結(jié)果插入到不同的存儲中。這個過程中就會出現(xiàn)一個問題，比如同一個中間結(jié)果，我們把不同的部分插入到不同的存儲中，那么就會有多條 DAG，雖然都是臨時結(jié)果，但是也會造成 Kafka Source 的重復消費，對性能和資源造成極大的浪費。于是我們想，是否可以避免臨時中間結(jié)果的多次消費。在 1.9 版本之前，我們進行了 StreamGraph 的重建，將三個 DataSource 的 DAG 進行了合并；在 1.9 版本，Magina 自己也提供了一個查詢和 Source 合并的優(yōu)化；但是我們發(fā)現(xiàn)如果是在同一個 data update 中有對同一個表的多個 Source 的引用，它自己會合并，但是如果不是在同一個 data update 中，是不會立即合并的，于是在 1.9 版本之后中我們對 modifyOperations 做了一個 buffer 來解決這個問題。

(二)同交換機流量激增消費計算延遲問題這個問題是最近才出現(xiàn)的問題，也可能不僅僅是同交換機，同機房的情況也可能。在同一個交換機下我們部署了很多機器，一部分機器部署了 Kafka 集群，還有一部分部署了 Hadoop 集群。在 Hadoop 上面我們可能會進行 Spark、Hive 的離線計算以及 Flink 的實時計算，Flink 也會消費 Kafka 進行實時計算。在運行的過程中我們發(fā)現(xiàn)某一個任務會出現(xiàn)整體延遲的情況，排查過后沒有發(fā)現(xiàn)其他的異常，除了交換機在某一個時間點的瀏覽激增，進一步排查發(fā)現(xiàn)是離線計算的瀏覽激增，又因為同一個交換機的帶寬限制，影響到了 Flink 的實時計算。

為解決這個問題，我們就考慮要避免離線集群和實時集群的相互影響，去做交換機部署或者機器部署的優(yōu)化，比如離線集群單獨使用一個交換機，Kafka 和 Flink 集群也單獨使用一個交換機，從硬件層面保證兩者之間不會相互影響。

五、Q & A

Q1：Kafka 在實時數(shù)倉中的數(shù)據(jù)可靠嗎？A1：這個問題的答案更多取決于對數(shù)據(jù)準確性的定義，不同的標準可能得到不同的答案。自己首先要定義好數(shù)據(jù)在什么情況下是可靠的，另外要在處理過程中有一個很好的容錯機制。Q2：我們在學習的時候如何去學習這些企業(yè)中遇到的問題？如何去積累這些問題？A2：個人認為學習的過程是問題推動，遇到了問題去思考解決它，在解決的過程中去積累經(jīng)驗和自己的不足之處。Q3：你們在處理 Kafka 的過程中，異常的數(shù)據(jù)怎么處理，有檢測機制嗎？A3：在運行的過程中我們有一個分發(fā)的服務，在分發(fā)的過程中我們會根據(jù)一定的規(guī)則來檢測哪些數(shù)據(jù)是異常的，哪些是正常的，然后將異常的數(shù)據(jù)單獨分發(fā)到一個異常的 Topic 中去做查詢等，后期用戶在使用的過程中可以根據(jù)相關(guān)指標和關(guān)鍵詞到異常的 Topic 中去查看這些數(shù)據(jù)。

---

? Flink?Forward?Asia 2020??官網(wǎng)上線啦洞察先機，智見未來，?Flink Forward Asia 2020 盛大開啟！誠邀開源社區(qū)的各方力量與我們一起，探討新型數(shù)字化技術(shù)下的未來趨勢，共同打造 2020 年大數(shù)據(jù)領(lǐng)域的這場頂級盛會！大會官網(wǎng)已上線，點擊「閱讀原文」即可預約峰會報名～(點擊可了解更多議題投遞詳情)戳我報名！

總結(jié)

以上是生活随笔為你收集整理的flink 写kafka_网易云音乐基于 Flink + Kafka 的实时数仓建设实践的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。