Flink 和 Pulsar 的批流融合

作者：由 Flink 中文社群發表于書法時間：2021-05-26

簡介：StreamNative 聯合創始人翟佳在本次演講中介紹了下一代雲原生訊息流平臺 Apache Pulsar，並講解如何透過 Apache Pulsar 原生的儲存計算分離的架構提供批流融合的基礎，以及 Apache Pulsar 如何與 Flink 結合，實現批流一體的計算。

Apache Pulsar 相對比較新，它於 2017 年加入 Apache 軟體基金會，2018 年才從 Apache 軟體基金會畢業併成為一個頂級專案。Pulsar 由於原生採用了儲存計算分離的架構，並且有專門為訊息和流設計的儲存引擎 BookKeeper，結合 Pulsar 本身的企業級特性，得到了越來越多開發者的關注。今天的分享分為 3 個部分：

Apache Pulsar 是什麼；

Pulsar 的資料檢視；

Pulsar 與 Flink 的批流融合。

一、Apache Pulsar 是什麼

下圖是屬於訊息領域的開源工具，從事訊息或者基礎設施的開發者對這些一定不會陌生。雖然 Pulsar 在 2012 年開始開發，直到 2016 年才開源，但它在跟大家見面之前已經在

雅虎

的線上運行了很長時間。這也是為什麼它一開源就得到了很多開發者關注的原因，它已經是一個經過線上檢驗的系統。

Pulsar 跟其他訊息系統最根本的不同在於兩個方面：

一方面，Pulsar 採用儲存計算分離的

雲原生架構

；

另一方面，Pulsar 有專門為訊息而設計的儲存引擎，Apache BookKeeper。

架構

下圖展示了 Pulsar 儲存

計算分離

的架構：

首先在計算層，Pulsar Broker 不儲存任何狀態資料、不做任何資料儲存，我們也稱之為服務層。

其次，Pulsar 擁有一個專門為訊息和流設計的儲存引擎 BookKeeper，我們也稱之為資料層。

這個分層的架構對使用者的叢集擴充套件十分方便：

如果想要支援更多的 Producer 和 Consumer，可以擴充上面無狀態的 Broker 層；

如果要做更多的資料儲存，可以單獨擴充底層儲存層。

這個雲原生的架構有兩個主要特點：

第一個是儲存計算的分離；

另外一個特點是每一層都是一個

節點對等

的架構。

從節點對等來說，Broker 層不儲存資料，所以很容易實現節點對等。但是 Pulsar 在底層的儲存也是節點對等狀態：在儲存層，BookKeeper 沒有采用 master/slave 這種主從同步的方式，而是透過 Quorum 的方式。

如果是要保持多個數據備份，使用者透過一個 broker 併發地寫三個儲存節點，每一份資料都是一個對等狀態，這樣在底層的節點也是一個對等的狀態，使用者要做底層節點的擴容和管理就會很容易。有這樣節點對等的基礎，會給使用者帶來很大的雲原生的便捷，方便使用者在每一層單獨擴容，也會提高使用者的線上系統的可用性和維護性。

同時，這種分層的架構為我們在 Flink 做批流融合打好了基礎。因為它原生分成了兩層，可以根據使用者的使用場景和批流的不同訪問模式，來提供兩套不同的 API。

如果是實時資料的訪問，可以透過上層 Broker 提供的 Consumer 介面；

如果是歷史資料的訪問，可以跳過 Broker，用儲存層的 reader 介面，直接訪問底層儲存層。

儲存 BookKeeper

Pulsar 另一個優勢是有專門為流和訊息設計的儲存引擎 Apache BookKeeper。它是一個簡單的 write-ahead-log 抽象。Log 抽象和流的抽象類似，所有的資料都是源源不斷地從尾部直接追加。

它給使用者帶來的好處就是寫入模式比較簡單，可以帶來比較高的吞吐。在一致性方面，BookKeeper 結合了 PAXOS 和 ZooKeeper ZAB 這兩種協議。BookKeeper 暴露給大家的就是一個 log 抽象。你可以簡單認為它的一致性很高，可以實現類似 Raft 的 log 層儲存。BookKeeper 的誕生是為了服務我們在 HDFS naming node 的 HA，這種場景對一致性要求特別高。這也是為什麼在很多關鍵性的場景裡，大家會選擇 Pulsar 和 BookKeeper 做儲存的原因。

BookKeeper 的設計中，有專門的

讀寫隔離

，簡單理解就是，讀和寫是發生在不同的磁碟。這樣的好處是在批流融合的場景可以減少與歷史資料讀取的相互干擾，很多時候使用者讀最新的實時資料時，不可避免會讀到歷史資料，如果有一個專門為歷史資料而準備的單獨的磁碟，歷史資料和實時資料的讀寫不會有 IO 的爭搶，會對批流融合的 IO 服務帶來更好的體驗。

應用場景

Pulsar 場景應用廣泛。下面是 Pulsar 常見的幾種應用場景：

第一，因為 Pulsar 有 BookKeeper，資料一致性特別高，Pulsar 可以用在計費平臺、支付平臺和交易系統等，對資料服務質量，一致性和可用性要求很高的場景。

第二種應用場景是 Worker Queue / Push Notifications / Task Queue，主要是為了實現系統之間的相互解耦。

第三種場景，與 Pulsar 對訊息和佇列兩種場景的支援比較相關。Pulsar 支援 Queue 消費模式，也支援 Kafka 高頻寬的消費模型。後面我會專門講解 Queue 消費模型與 Flink 結合的優勢。

第四個場景是 IoT 應用，因為 Pulsar 在服務端有 MQTT 協議的解析，以及輕量級的計算 Pulsar Functions。

第五個方面是 unified data processing，把 Pulsar 作為一個批流融合的儲存的基礎。

我們在 2020 年 11 月底的 Pulsar Summit

亞洲峰會

，邀請 40 多位講師來分享他們的 Pulsar 落地案例。如果大家對 Pulsar 應用場景比較感興趣，可以關注 B 站上 StreamNative 的賬號，觀看相關影片。

二、Pulsar 的資料檢視

在這些應用場景中，Unified Data Processing 尤為重要。關於批流融合，很多國內使用者的第一反應是選擇 Flink。我們來看 Pulsar 和 Flink 結合有什麼樣的優勢？為什麼使用者會選擇 Pulsar 和 Flink 做批流融合。

首先，我們先從 Pulsar 的資料檢視來展開。跟其他的訊息系統一樣，Pulsar 也是以訊息為主體，以 Topic 為中心。所有的資料都是 producer 交給 topic，然後 consumer 從 topic 訂閱消費訊息。

Partition 分割槽

為了方便擴充套件，Pulsar 在 topic 內部也有分割槽的概念，這跟很多訊息系統都類似。上面提到 Pulsar 是一個分層的架構，它採用分割槽把 topic 暴露給使用者，但是在內部，實際上每一個分割槽又可以按照使用者指定的時間或者大小切成一個分片。一個 Topic 最開始建立的時候只有一個 active 分片，隨著使用者指定的時間到達以後，會再切一個新的分片。在新開一個分片的過程中，儲存層可以根據各個節點的容量，選擇容量最多的節點來儲存這個新的分片。

這樣的好處是，topic 的每一個分片都會均勻地散佈在儲存層的各個節點上，實現資料儲存的均衡。如果使用者願意，就可以用整個儲存叢集來儲存分割槽，不再被單個節點容量所限制。如下圖所示，該 Topic 有 4 個分割槽，每一個分割槽被拆成多個分片，使用者可以按照時間（比如 10 分鐘或者一個小時），也可以按照大小（比如 1G 或者 2G）切一個分片。分片本身有順序性，按照 ID 逐漸遞增，分片內部所有訊息按照 ID 單調遞增，這樣很容易保證順序性。

Stream 流儲存

我們再從單個分片來看一下，在常見流（stream）資料處理的概念。使用者所有的資料都是從流的尾部不斷追加，跟流的概念相似，Pulsar 中 Topic 的新資料不斷的新增在 Topic 的最尾部。不同的是，Pulsar 的 Topic 抽象提供了一些優勢：

首先，它採用了儲存和計算分離的架構。在計算層，它更多的是一個訊息服務層，可以快速地透過 consumer 介面，把最新的資料返回給使用者，使用者可以實時的獲取最新的資料；

另外一個好處是，它分成多個分片，如果使用者指定時間，從元資料可以找到對應的分片，使用者可以繞過實時的流直接讀取儲存層的分片；

還有一個優勢是，Pulsar 可以提供無限的流儲存。

做基礎設施的同學，如果看到按照時間分片的架構，很容易想到把老的分片搬到二級儲存裡面去，在 Pulsar 裡也是這樣做的。使用者可以根據 topic 的消費熱度，設定把老的，或者超過時限或大小的資料自動搬到二級儲存中。使用者可以選擇使用 Google，微軟的 Azure 或者 AWS 來儲存老的分片，同時也支援 HDFS 儲存。

這樣的好處是：對最新的資料可以透過 BookKeeper 做快速返回，對於老的冷資料可以利用網路儲存雲資源做一個無限的流儲存。這就是 Pulsar 可以支援

無限流儲存

的原因，也是批流融合的一個基礎。

總體來說，Pulsar 透過儲存計算分離，為大家提供了實時資料和歷史資料兩套不同的訪問介面。使用者可以依據內部不同的分片位置，根據 metadata 來選擇使用哪種介面來訪問資料。同時根據分片機制可以把老的分片放到二級儲存中，這樣可以支撐無限的流儲存。

Pulsar 的統一體現在對分片元資料管理的方面。每個分片可以按照時間存放成不同的儲存介質或格式，但 Pulsar 透過對每個分片的 metadata 管理，來對外提供一個分割槽的邏輯概念。在訪問分割槽中的一個分片的時候我可以拿到它的元資料，知道它的在分割槽中的順序，資料的存放位置和儲存型別 Pulsar 對每一個分片的 metadata 的管理，提供了統一的 topic 的抽象。

三、Pulsar 和 Flink 的批流融合

在 Flink 中，流是一個基礎的概念，Pulsar 可以作為流的載體來儲存資料。如果使用者做一個批的計算，可以認為它是一個有界的流。對 Pulsar 來說，這就是一個 Topic 有界範圍內的分片。

在圖中我們可以看到，topic 有很多的分片，如果確定了起止的時間，使用者就可以根據這個時間來確定要讀取的分片範圍。對實時的資料，對應的是一個連續的查詢或訪問。對 Pulsar 的場景來說就是不停的去消費 Topic 的尾部資料。這樣，Pulsar 的 Topic 的模型就可以和 Flink 流的概念很好的結合，Pulsar 可以作為 Flink 流計算的載體。

有界的計算可以視為一個有界的流，對應 Pulsar 一些限定的分片；

實時的計算就是一個無界的流，對 Topic 中最新的資料做查詢和訪問。

對有界的流和無界的流，Pulsar 採取不同的響應模式：

第一種是對歷史資料的響應。如下圖所示，左下角是使用者的 query，給定起止的時間限定流的範圍。對 Pulsar 的響應分為幾步：

第一步，找到 Topic，根據我們統一管理的 metadata，可以獲取這個 topic 裡面所有分片的 metadata 的列表；

第二步，根據時間限定在 metadata 列表中，透過兩分查詢的方式來獲取起始分片和終止的分片，選擇需要掃的分片；

第三步，找到這些分片以後透過底層儲存層的介面訪問需要訪問的這些分片，完成一次歷史資料的查詢。

對實時資料的查詢，Pulsar 也提供和 Kafka 相同的介面，可以透過 consumer 的方式來讀取最尾端分片（也就是最新的資料），透過 consumer 介面對資料進行實時訪問。它不停地查詢最新的資料，完成之後再進行下一次查詢。這種情況下，使用 Pulsar Pub/Sub 介面是一種最直接最有效的方式。

簡單來說，Flink 提供了統一的檢視讓使用者可以用統一的 API 來處理 streaming 和歷史資料。以前，

資料科學家

可能需要編寫兩套應用分別用來處理實時資料和歷史資料，現在只需要一套模型就能夠解決這種問題。

Pulsar 主要提供一個數據的載體，透過基於分割槽分片的架構為上面的計算層提供流的儲存載體。因為 Pulsar 採用了分層分片的架構，它有針對流的最新資料訪問介面，也有針對批的對併發有更高要求的儲存層訪問介面。同時它提供無限的流儲存和統一的消費模型。

四、Pulsar 現有能力和進展

最後我們額外說一下 Pulsar 現在有怎樣的能力和最近的一些進展。

現有能力

schema

在大資料中，schema 是一個特別重要的抽象。在訊息領域裡面也是一樣，在 Pulsar 中，如果 producer 和 consumer 可以透過 schema 來簽訂一套協議，那就不需要生產端和消費端的使用者再線下溝通資料的傳送和接收的格式。在

計算引擎

中我們也需要同樣的支援。

在 Pulsar-Flink connector 中，我們借用 Flink schema 的 interface，對接 Pulsar 自帶的 Schema，Flink 能夠直接解析儲存在Pulsar 資料的 schema。這個 schema 包括兩種：

第一種是我們常見的對每一個訊息的元資料（meatdata）包括訊息的 key、訊息產生時間、或是其他元資料的資訊。

另一種是對訊息的內容的

資料結構

的描述，常見的是 Avro 格式，在使用者訪問的時候就可以透過Schema知道每個訊息對應的資料結構。

同時我們結合

Flip-107

，整合 Flink

metadata schema

和 Avro 的 metadata，可以將兩種 Schema 結合在一起做更復雜的查詢。

source

有了這個 schema，使用者可以很容易地把它作為一個 source，因為它可以從 schema 的資訊理解每個訊息。

Pulsar Sink

我們也可以把在 Flink 中的計算結果返回給 Pulsar 把它做為 Sink。

Streaming Tables

有了 Sink 和 Source 的支援，我們就可以把 Flink table 直接暴露給使用者。使用者可以很簡單的把 Pulsar 作為 Flink 的一個 table，查詢資料。

write to straming tables

下圖展示如何把計算結果或資料寫到 Pulsar 的 Topic 中去。

Pulsar Catalog

Pulsar 自帶了很多企業流的特性。Pulsar 的 topic（e。g。 persistent：//tenant_name/namespace_name/topic_name）不是一個平鋪的概念，而是分很多級別。有 tenant 級別，還有 namespace 級別。這樣可以很容易得與 Flink 常用的 Catalog 概念結合。

如下圖所示，定義了一個 Pulsar Catalog，database 是 tn/ns，這是一個路徑表達，先是 tenant，然後是 namespace，最後再掛一個 topic。這樣就可以把Pulsar 的 namespace 當作 Flink 的 Catalog，namespace 下面會有很多 topic，每個 topic 都可以是 Catalog 的 table。這就可以很容易地跟 Flink Cataglog 做很好的對應。在下圖中，上方的是 Catalog 的定義，下方則演示如何使用這個 Catalog。不過，這裡還需要進一步完善，後邊也有計劃做 partition 的支援。

FLIP-27

是 Pulsar - Flink 批流融合的一個代表。前面介紹了 Pulsar 提供統一的檢視，管理所有 topic 的 metadata。在這個檢視中，根據 metadata 標記每個分片的資訊，再依靠 FLIP-27 的 framework 達到批流融合的目的。FLIP-27 中有兩個概念：Splitter 和 reader。

它的工作原理是這樣的，首先會有一個 splitter 把資料來源做切割，之後交給 reader 讀取資料。對 Pulsar 來說，splitter 處理的還是 Pulsar 的一個 topic。抓到 Pulsar topic 的 metadata 之後，根據每個分片的元資料來判斷這個分片儲存在什麼位置，再選最合適的 reader 進行訪問。Pulsar 提供統一的儲存層，Flink 根據 splitter 對每個分割槽的不同位置和格式的資訊，選擇不同的 reader 讀取 Pulsar 中的資料。

Source 高併發

另一個和 Pulsar 消費模式緊密相關的是。很多 Flink 使用者面臨的問題是如何讓 Flink 更快地執行任務。例如，使用者給了 10 個併發度，它會有 10 個 job 併發，但假如一個 Kafka 的 topic 只有 5 個分割槽，由於每個分割槽只能被一個 job 消費，就會有 5 個 Flink job 是空閒的。如果想要加快消費的併發度，只能跟業務方協調多開幾個分割槽。這樣的話，從消費端到生產端和後邊的運維方都會覺得特別複雜。並且它很難做到實時的按需更新。

而 Pulsar 不僅支援 Kafka 這種每個分割槽只能被一個 active 的 consumer 消費的情況，也支援 Key-Shared 的模式，多個 consumer 可以共同對一個分割槽進行消費，同時保證每個 key 的訊息只發給一個 consumer，這樣就保證了 consumer 的併發，又同時保證了訊息的有序。

對前面的場景，我們在 Pulsar Flink 裡做了 Key-shared 消費模式的支援。同樣是 5 個分割槽，10 個併發 Flink job。但是我可以把 key 的範圍拆成 10 個。每一個 Flink 的子任務消費在 10 個 key 範圍中的一個。這樣從使用者消費端來說，就可以很好解耦分割槽的數量和 Flink 併發度之間的關係，也可以更好提供資料的併發。