借助Flink與Pulsar，BIGO打造實時消息處理系統(tǒng)

簡介：BIGO 從 Kafka 轉向 Pulsar，并借助 Apache Pulsar 和 Flink 構造實時消息流處理系統(tǒng)。

本文整理自 BIGO Staff Engineer 陳航在 Flink Forward Asia 2020 分享的議題《借助 Flink 與 Pulsar，BIGO 打造實時消息處理系統(tǒng)》。主要內(nèi)容包括： 關于 BIGO BIGO 為什么會選擇 Apache Pulsar Apache Pulsar 在 BIGO 中的角色 BIGO 借助 Apache Pulsar 和 Flink 構造自己的實時消息流處理系統(tǒng)。 

一、關于 BIGO

借助于大數(shù)據(jù)和人工智能技術，BIGO 基于視頻的服務和產(chǎn)品獲得了廣泛的歡迎，在 150 多個國家和地區(qū)獲得了大量的用戶。BIGO 主要有兩款非常流行的產(chǎn)品，第一款是 BIGO Live，另外一款是 Likee。BIGO Live 是一個直播平臺，而 Likee 是一個短視頻平臺。

二、為什么選擇 Apache Pulsar

在過去的幾年里，BIGO 的消息平臺主要還是以開源的 Kafka 集群為主，但是隨著業(yè)務的不斷增長、用戶不斷擴充，整個消息流平臺所承載的消息量和數(shù)據(jù)量也出現(xiàn)了成倍的增長，同時也對整個消息流系統(tǒng)提出了更高的要求。

主要體現(xiàn)在以下幾個方面：

• 第一，它對整個系統(tǒng)的穩(wěn)定性、可擴展性以及魯棒性提出了更高的要求。

• 第二，由于我們是短視頻推薦相關的服務，所以對整個消息流的低延遲也提出了非常高的要求。

隨著數(shù)量增長，整個 BIGO 的消息流平臺的團隊在維護多個 Kafka 集群上付出了大量的工作，擺在我們面前的有很多 Kafka 集群運維相關的問題。這個時候，我們就在思考，我們是選擇開源 Kafka 的一個基線版本進行自己的迭代開發(fā)呢？還是看一下開源社區(qū)里面有哪些可以借鑒的方案，來打造一個符合我們應用場景需求的消息流平臺。

于是我們進行了一系列調(diào)研。在調(diào)研的過程中，我們的目光注意到了 Apache Pulsar，它有以下幾點 feature 比較 match 我們的應用場景：

• 首先，它能夠水平地擴展。我們知道對于 Kafka 而言，它是一個服務和存儲綁定的系統(tǒng)。當我們需要去擴容一個集群的時候，單單把機器上線是不能夠滿足需求的，我們需要對整個 topic 的 partition 進行相應操作，這個時候就是耗人力去運維的。所以，我們需要有一個能夠水平擴展的系統(tǒng)。而 Apache Pulsar 提供的是存儲和服務分離的一個架構，使用的是 bookkeeper 作為底層的數(shù)據(jù)存儲，上層有一個 broker 來提供相關的服務。

• 另外，就是它的 low latency 還有高吞吐、低延遲以及在雅虎的生產(chǎn)環(huán)境上面經(jīng)受了大數(shù)據(jù)量的考驗。

• 跨集群的復制等一系列的 feature 對于我們而言也是非常需要的。

• 并且，這樣一個存儲和服務分離的架構也極大地減少了人工運維的成本。

所以我們選擇了 Apache Pulsar。

三、Apache Pulsar 在 BIGO 中的角色1. 引入 Pulsar 的歷程

在 2019 年 11 月，我們重新開始思考 BIGO 的應用場景下面所需要的消息流平臺到底是什么樣的。是基于一個開源的 Kakfa 框架去開發(fā)，還是選擇另外一套整個消息流系統(tǒng)？

在 2019 年 11 月份，我們做了一次整個消息流平臺的調(diào)研工作。在調(diào)研過程中，我們對比了 Kafka、RocketMQ、Apache Pulsar 等業(yè)界相近的相對的消息隊列的實現(xiàn)。然后我們做了一系列的橫向?qū)Ρ龋⑶腋覀兊臉I(yè)務需求進行了相應的比較。最終發(fā)現(xiàn)使用 Apache Pulsar 能夠解決我們生產(chǎn)上的一些問題，能夠為我們的消息流平臺提供非常好的運維相關的負擔的減輕，以及整個系統(tǒng)的穩(wěn)定性和吞吐的提升，所以我們就選擇了 Apache Pulsar。

在 2019 年 12 月份，我們進行了一系列的壓測。任何一個開源的框架，如果沒有經(jīng)過公司內(nèi)部的大流量場景下的壓測，是不敢上線的。所以從 2019 年 12 月份一直到 2020 年 4 月份，經(jīng)過了一系列的長時間的壓測工作。

在壓測的過程中，我們同時也發(fā)現(xiàn)了 Apache Pulsar 的一些問題，并且給社區(qū)修了一系列的 bug。在 2020 年 4 月份，我們把 Apache Pulsar 部署在了我們的生產(chǎn)測試環(huán)境；在穩(wěn)定運行一個月之后，我們就把它部署到了生產(chǎn)環(huán)境；在 2020 年 5 月份，正式上線。

現(xiàn)有的 Apache Pulsar 集群規(guī)模，目前有十幾個 Apache Pulsar 的節(jié)點。整個集群的入流量是在 2~3 GB/s。隨著時間的推移，也有越來越多的應用會不斷地遷移到 Apache Pulsar 來替代現(xiàn)有的 Kafka 集群。

2. Apache Pulsar 的角色

Apache Pulsar 在整個流處理過程中提供的是一個 PUB-SUB 的角色。

• 首先，有 BIGO 這邊的 Baina，一個 C++ 實現(xiàn)的消息收集服務，把相關的數(shù)據(jù)寫到 Apache Pulsar 相關的 topic 里面去，這是第一種場景。

• 第二場景就是 KMM，也就是 Kafka 的 Mirror Maker。

• 第三種場景是 Flink。另外就是一些各種語言的客戶端所實現(xiàn)的 producer。它的下游主要有 Flink、Flink SQL 以及各個語言所實現(xiàn)的 consumer，比如說 golang、JAVA，C++ 的等等。

3. 下游支撐的業(yè)務場景

第一個是實時數(shù)倉，第二個是實時的 ETL，第三個是實時數(shù)據(jù)分析，另外就是實時推薦，還有更多的業(yè)務場景也在逐漸的介入。下游的數(shù)據(jù)會寫到 Hive、Pulsar 的 topic、ClickHouse、Hadoop、redis 等一系列下游的相關存儲系統(tǒng)。

四、Apache Pulsar 和 Flink 構造實時消息流處理系統(tǒng)。

這里需要分為以下三個方面來講：

• 第一，是關于 Pulsar-Flink-Connector 的一些內(nèi)幕。我相信在介紹 Pulsar-Flink-Connector 的一些內(nèi)幕之后，大家會對整個 Flink 與 Pulsar 之間結合的關系會更加地清晰明亮，在使用過程中也會更加地清晰；

• 第二，是 BIGO 的一個 use case，就是使用 Apache Pulsar 和 Flink 來打造自己的實時 ETL 處理系統(tǒng)；

• 第三，是借助 Apache Pulsar 和 Flink 打造 AB-test 系統(tǒng)。

首先看一下 Pulsar-Flink-Connector 整個生產(chǎn)和消費的邏輯。它主要包括一個 source 的 API 和 sink 的 API。對于消費的時候，也就是使用一個 Pulsar-Flink-Connector 的 source 來訂閱 Pulsar 的一個 topic。另外一個就是我們寫一個 sink，會把 Flink 里面的數(shù)據(jù)寫出到 Pulsar 的 topic 里面。下圖左邊的代碼展示怎么去訂閱這樣一個 topic，實際上只需要 new 一個 FlinkPulsarSource 的一個流，然后把這條流加入到 DataStream 里面去就可以了。

對于 Flink 數(shù)據(jù)的寫出而言，只需要 new 一個 FlinkPulsar 的 Sink，然后我們調(diào)用第二個 DataStream 的 sink 就可以把數(shù)據(jù)給寫出去了。實際上，整個的實現(xiàn)而言，跟 Kafka 的 API 是非常類似的。這里需要注意的幾點就是，對于 FlinkPulsarSource 里面需要傳入的是 serviceUrl 以及 adminUrl。

• serviceUrl 類似于 Kafka 的 broker_list；

• adminUrl 就是我們?nèi)ヒ怨芾韱T的方式來控制 Pulsar 的一些相關的操作。

Pulsar Flink 怎么樣來訂閱 Pulsar 的 topic，怎么樣消費以及它的 offset 是怎么樣 commit 回去的？

這里就會涉及到 Pulsar Flink 的 exactly-once source。咱們首先來看一下圖左邊部分。這個圖里面有一個 Pulsar 的 topic，當我們 new 一個 PulsarFlinkSource 的時候，實際上會對每一個 Pulsar topic 的 partition 創(chuàng)建一個 reader。這個 reader 使用的是 Non-Durable Cursor，當這個 reader 訂閱了這個 topic 之后，這個 topic 的數(shù)據(jù)流就會源源不斷地流到這個 reader 的線程里面去。當 reader 的線程觸發(fā)一次 checkpoint 的時候，這個 Flink 任務就會把自己的一些狀態(tài) checkpoint 起來。當 checkpoint 完成的時候，就會調(diào)用一次 Notify checkpoint complete 這樣的一個方法。觸發(fā)的是另外一個 subscription 的一個 commit。

這個 subscription 實際上是一個 durable cursor。當它 commit offset 的時候，這個 offset 會保存在 bookkeeper 里面，這是一個永久保存的 offset。這樣做的好處是，當 checkpoint 失敗或者 checkpoint 丟了的時候，我們需要以一個 subscription name 從 Pulsar 里面去恢復的時候，就可以從 bookkeeper 里面去把 message id 讀出來，然后從這邊恢復。

實際上對于 Pulsar-Flink-Connector 的消費而言，它是由一條數(shù)據(jù)流和一條控制流來組成的：

• 對于數(shù)據(jù)流，就是 Pulsar Topic 的數(shù)據(jù)源源不斷的會流入到 reader 的這樣一個線程里面，由 reader 線程進行相應的處理。

• 控制流就是通過 subscription name 來提交消費的 message id，也就是類似于 Kafka 的一個 offset，會提交到 Pulsar 的客戶端來保證消費的位置。

接下來看一下 checkpoint 實現(xiàn)的流程。

• 首先，當我們?nèi)プ?checkpoint N 的時候，當 N 結束了之后，它會把 durable cursor 進行一次 commit；

• 當我們?nèi)プ?checkpoint N+1 的時候，N+1 完成之后，會接著去 commit N+1 的 durable cursor。

這樣做的好處是，當這個任務失敗之后，如果需要從某一個 checkpoint 恢復，只需要從 checkpoint 里面去讀到上一次 checkpoint 成功的 offset 的 durable cursor 的 message id 的位置，就可以從上一次的位置去消費了。這也是保證 source 的 exactly once 的實現(xiàn)。

Topic/Partition 的 Discovery

• 第一點是，在 Pulsar-Flink-Connector 實現(xiàn)的邏輯里，會為每一個 Topic/Partition 分配一個 reader 的線程。

• 第二點是，每一個 task manager 會包括多個 reader 的線程，這地方會有一個什么樣的映射關系？

舉個例子：假設我們訂閱的 Topic 里面，有 10 個 partition，F(xiàn)link 里面只給它分配 5 個 task manager，那么怎么將 partition 映射到 5 個 task manager 里面去？這就會涉及到一個分配的邏輯。整個分配的邏輯，實際上是使用一個哈希的方式把某一個 Topic/Partition hash 到目標的 task manager 上面。

這就會存在一些隱患：當我們訂閱了幾百個甚至上千個 topic 的時候，可能會存在一定的分配不均衡。成百上千個 Topic/Partition 里面，并不是每一個 partition 的流量都是均衡的。假設我們訂閱了十個 Topic，其中有九個 Topic 的流量很小，另外一個 Topic 的流量很大，那么均攤到某一個 partition 時候也是這樣的。這個很大的 topic 的 Partition 的流量很大，另外 Topic/Partition 的流量很小。如果我們只是單純地進行一次 hash 的話，就會造成某些 task manager 上面的流量不均衡，可能會導致頻繁 GC 的問題。這個問題在下一個 use case 里會詳細地提到，以及怎么樣去解決它。

另外就是當某一個 Topic/Partition 進行一次分區(qū)擴容時，怎么樣去自動訂閱這樣一個新的分區(qū)？在 Pulsar-Flink-Connector 里面會啟動一個定時 check 的線程的邏輯。假設我們每一分鐘 check 一次，是否有新的 partition 的加入，并且這個新 Topic/Partition 分配到了某一個 task manager 上面，那么這個 task manager 就會自動地新創(chuàng)建一個 reader 的線程，然后把這個 partition 訂閱下來。

這整個的流程，會有一個 discover 會不斷的去 check。當有新的 partition 的時候就會 new 一個 reader 起來。每一個 reader 獨立消費某一個 Topic/Partition，把數(shù)據(jù)拿過來之后會定期進行自己的反序列化操作以及后續(xù)的處理。

上面講到的是整個 connector 的一個邏輯。在 Pulsar-Flink-Connector 里面提供了 job 的方式，還提供了 catalog 的方式來消費 Pulsar 的 topic。但是目前它是沒有提供 SQL DDL 的方式，在 BIGO 的應用場景里面大部分的 topic 都是以 json 的格式。大部分的數(shù)據(jù)，都是以 json 格式寫入的。

對于這一類 json 格式的topic，它可能沒有事先配置自己的 schema 就寫進來了，那么我們在消費的時候，如果想用 SQL，怎么辦呢？這里就需要用到 Flink DDL 的框架，所以 BIGO 的消息流平臺團隊在我們的使用過程中為 Pulsar-Flink-Connector 開發(fā)了 Flink SQL DDL 的支持。接下來看一下 Flink SQL DDL 的框架。

• 第一步，圖左邊就是 fetch message from Pulsar topic，首先會定義這個 topic 的里面數(shù)據(jù)的一個字段信息，也就是 create table test_Flink_SQL，這里面有 rid 等字段。下面的位置里面包含的是怎樣去和 Pulsar 的服務端建立連接的，這里會指定 topic 名稱，指定 service url，admin url 以及 subscribe name，還有一些一系列相關的配置操作。這樣一段 SQL 的代碼就能夠很好地完成把數(shù)據(jù)從 Pulsar topic 里面給消費出來。

• 第二步，就可以進行一系列應用層相關邏輯的處理。比如做 join，count、union 等操作。另外就是一些應用層邏輯的處理，比如說去做統(tǒng)計相關的一些操作。在第二步操作完了之后，我們需要將最終的結果寫出到第三方存儲里面。第三方存儲會包括 Hive 表、HDFS 和 Pulsar 的 topic 等。

• 對于最終的寫入寫出就會進入到第三步，我們會調(diào)用一個 insert into 的方法，直接把我們處理的結果，寫出到相關的 Hive 表里面去，這就是整個 Flink SQL DDL 的一個處理邏輯。我們借助 Flink SQL DDL 能夠很好地來實現(xiàn)我們的 AB test 相關的操作。那么在前面的講解里面，我們可能會使用一個 job 的方式來提交，有了 Flink SQL DDL 的支持，我們就可以很方便地使用一個 SQL 的方式來消費 Pulsar 的 topic，會進行一系列邏輯處理，最終把結果寫出去。

現(xiàn)在來看一下基于 SQL 方式的 use case。

Case 1

首先來看一下 BIGO reall-time ETL 的實現(xiàn)。這個實時 ETL 的背景，是我們在 Pulsar 里面，會有成百上千個 topic，每一個 topic 會有自己獨立的 schema。我們現(xiàn)在的一個需求是想要把每一個 topic 使用自己的 schema 進行一次解析，把最終解析的結果以 bucket 的格式落到 HDFS 的 Hive 表上面去。對于這樣一個需求，我們可能會有幾種方案：

• 第一種方案，我們會直接使用 Pulsar 的 HDFS 的 connector，會把 topic 里面的數(shù)據(jù)會消費出來然后落到 HDFS 上面去，這樣做的話，當我們需要對 topic 里面進行一系列的處理的時候，可能就不大好辦了。另外一個就是我們有成百上千個 topic，那么也會有成百上千個 schema，也就是說我們可能要維護成百上千個線程，去解相應的 topic 里面的數(shù)據(jù)，然后把它落出去。這樣對于整個任務的維護成本可能會比較高。

• 第二種方案。我們可以用 Flink SQL 去消費每個topic，每一個 SQL 指令自己的 schema，然后把這個 topic 給消費出來，之后進行一系列的處理，然后寫出去。這種方式，實際上也會帶來幾百個甚至上千個 SQL 任務的維護工作。

• 第三個方案，我們想到了使用一個 Flink 任務來消費成百上千個 Pulsar 的 topic。然后進行一系列的 ETL 處理，首先進行 schema 的解析，然后進行一系列邏輯處理，最終把它寫出到 HDFS 上面去。下面這張圖，就是我們采用的第三種方案：使用一個 Flink 的 job 把成百上千個 topic 訂閱了。訂閱完了之后，獲取相應的線程去消費。解析完了之后會經(jīng)過一系列邏輯處理，最終顯示到 HDFS 上面去。

這個 case 可能存在數(shù)據(jù)分布不均的問題。假設，我們有 500 個 topic，其中 400 個 topic 的流量很小，另外 100 個 topic 的流量很大。那么我們在訂閱的時候，假設我起了 100 個 task manager 去消費。那么這可能就會按平均來算，有 5-10 個 topic partition 會落到同一個 task manager 上面去。如果我們不干預的話，由于這個 partition 自身的流量不均衡，可能會導致它從運行任務的進程的流量也是不均衡的，帶來了頻繁 GC 的問題。

為了解決消費端上面的 task manager 流量不均衡的情況。我們引入了一個 slot group 的概念。我們會事先對 topic partition 的流量進行一個預估，預估完了之后，會通過人工計算的方式把幾個低流量的 topic 組合成同一個 group 里面。同一個 group 的 topic 會被放在同一個 slot 里面，然后去進行調(diào)度，這樣就能夠很好的去把 task manager 上面的消費流量不均的問題解決掉了，整個 Flink job 就會運行的很好。

Case 2

第二個 case 是一個 AB test 的應用場景，做這個 AB test 場景的一個初衷是什么呢？在我們實時的數(shù)倉里面，需要去產(chǎn)出小時級別的中間表，以及天級的中間表，給推薦算法的工程師以及數(shù)據(jù)分析師來使用。對于小時級別的中間表以及天級的中間表的產(chǎn)生，需要通過實時的去計算底層的各種類型的打點，比如用戶觀看的打點、某個視頻的下發(fā)打點，還有用戶其他行為的打點等等，會按照某一個維度進行聚合。聚合了之后會進行相關的一些統(tǒng)計，最終會形成一張寬帶供推薦算法工程師以及數(shù)據(jù)分析師來使用。

如果我們不提供這樣一個寬表的話，那么對于上層的業(yè)務方而言，可能要不斷的去訪問底層的表，對底層表進行各種相應的操作。這樣不但會浪費數(shù)據(jù)分析師以及推薦算法工程師的時間，也會造成整個集群計算資源的浪費。那么在 BIGO 這邊，之前的解決方案是使用 Hive。使用 Map Reduce 的方式，來把每張底層的表進行一次聚合操作。聚合完了之后會提供一個小時級別中間表以及天級的中間表給上層業(yè)務使用，這樣做的弊端是：Hive Map Reduce 的時效性是沒法保證的。所以我們就在想能否使用 Flink 流式計算的方式來提高實時數(shù)倉的數(shù)據(jù)產(chǎn)出效率。

接下來就是我們這邊的一個解決方案：首先我們會用 Flink SQL 去消費 Pulsar 的 topic。從下圖的左邊來看，我們有 Topic A、Topic B 和 Topic K。每個 topic 有自己的 DDL。我們首先會使用 Flink SQL 加上每一個 topic 的 scanner，也就是 DDL 會把 topic 的數(shù)據(jù)從 Pulsar 里面加載出來，然后把它做成每個 topic 的一個視圖。

這個地方我們就會有 Table A、Table B 和 Table K 的一個表。假設有 K 張表，那么我們需要對 K 張表進行一次聚合操作。假設是按照 uid 進行一次聚合，那么這個聚合有兩種方式：

第一種方式是做 join。對于 Flink 而言，它的流式 join 可能耗時會比較長，整個計算資源的消耗也是非常大的。所以我們這邊做了一個比較巧妙的方案就是使用 union 代替 join。我們會把 Table A、Table B 和 Table K 通過 union 的方式會生成一個 View X。然后把 View X 直接寫出以小時為粒度，到 ClickHouse 供用戶查詢。在 union 的過程當中，我們還會做一些相關的聚合的操作。來把相關的指標給聚合起來供用戶使用。這個就是小時級別的中間表。

對于天級的中間表而言，我們所遇到的挑戰(zhàn)是：它并不是單單的只依賴了 Table A、Table B 和 Table K，可能還依賴了離線的表。假設有 Table a1、Table a2 和 Table a3 三張表。我們怎么樣把實時的表和離線的表做一個關聯(lián)？這里我們也是使用的一個比較巧妙的方式。

首先。在左邊 Table A、Table B 和 Table K 會使用 Flink SQL 把數(shù)據(jù)從 Pulsar 消費出來，然后做成一個獨立的 table。然后同樣也是以 union 的方式把實時的流表給 union 起來，做一些統(tǒng)計相關的處理生成一個視圖，一個View X。這個 View X 會根據(jù)我們精心設計過的一個 row-key，把它以天為維度寫出到 HBase 里面去。

另外，對于離線而言。因為我們 Table A、Table B 和 Table K 只是代表了咱們實時的一些數(shù)據(jù)，對于離線的數(shù)據(jù)，也是需要 join 進來的，那么就會使用一個 Spark 來把 Table a1、Table a2 和 Table a3 相關的數(shù)據(jù)給 join 起來，然后也以相同的規(guī)則生成一個 row-key 寫在 HBase 里面去。

對于 HBase 而言，它實際上提供的就是一個 join 操作，寫到 HBase 就很好的避免了我們將 View X 以及 Spark 所生成的這樣一張表做 join 了。因為如果是有相同的 key，那么假設 HBase 這樣一張寬表有 100 列，View X 占了前 80 列，那么后面的 Spark 所算出來的這個表會自動地填充到那個后 20 列里面去，那么最終會生成同一個 row-key 的一個 100 維的一張寬表。那么我們接下來會把 HBase 里面這樣一張寬表讀出來，然后寫到 ClickHouse 供上層用戶去查詢。這樣就能夠很好的去避免表之間的 join 操作，極大地提高 join 的效率。

五、未來工作

• 首先，我們會接著在 Pulsar-Flink-Connector 上面繼續(xù)的去開發(fā)新的 feature 并且持續(xù)的去進行一系列的 bug 修復；

• 第二點，我們會更多的將 Flink 任務持續(xù)地從 Kakfa 遷移到 Apache Pulsar 上面去；

• 第三點，在我們整個的消息流平臺里，之前使用的是 Kakfa，可能有成百上千個 Flink 的任務或者是其他的任務，使用 Kafka 的 API 來消費 Kafka 的 topic。如果不提供一個簡單的方式讓用戶來消費 Pulsar 的 topic 的話，這個遷移工作是非常難進行的。所以我們會借助于 KOP，也就是 Kakfa on Pulsar，方便上層應用的遷移，有了這樣一層 KOP 的一個 proxy，對于上面應用程序是不需要改任何的代碼就能夠自動的從 Kafka 切到 Pulsar 上面的；

• 第四點，我們打算實現(xiàn)一個批流統(tǒng)一的數(shù)據(jù)的消費，從 Pulsar topic 里面以批或者是流的方式來消費 topic 里的數(shù)據(jù)；

• 第五點，我們會持續(xù)加強 Pulsar 以及 bookkeeper 的穩(wěn)定性以及吞吐的調(diào)優(yōu)；

• 第六點，我們會持續(xù)的去優(yōu)化 Bookkeeper 的 IO 協(xié)議棧。