Kafka-on-Pulsar的開發(fā)歷程是怎樣的

這期內(nèi)容當(dāng)中小編將會給大家?guī)碛嘘P(guān)Kafka-on-Pulsar 的開發(fā)歷程是怎樣的，文章內(nèi)容豐富且以專業(yè)的角度為大家分析和敘述，閱讀完這篇文章希望大家可以有所收獲。

成都創(chuàng)新互聯(lián)專注于鳳泉企業(yè)網(wǎng)站建設(shè),響應(yīng)式網(wǎng)站建設(shè),成都做商城網(wǎng)站。鳳泉網(wǎng)站建設(shè)公司,為鳳泉等地區(qū)提供建站服務(wù)。全流程按需設(shè)計(jì)網(wǎng)站，專業(yè)設(shè)計(jì)，全程項(xiàng)目跟蹤，成都創(chuàng)新互聯(lián)專業(yè)和態(tài)度為您提供的服務(wù)

Protocol Handler是在 Pulsar 2.5.0 版本后加入的新機(jī)制，希望開發(fā)者們能利用 Pulsar 已有的基礎(chǔ)架構(gòu)，把 Pulsar 當(dāng)作一個可靠、高效、穩(wěn)定的流數(shù)據(jù)存儲，可以利用它去開發(fā)一些可插拔消息協(xié)議。

所以 Kafka-on-Pulsar是基于 Protocol Handler 進(jìn)行開發(fā)的，支持 Kafka 2.0 協(xié)議的插件。只需下載 KoP 插件并安裝到已有的 Pulsar broker 里，就能在 Pulsar 里支持 Kafka 協(xié)議。優(yōu)點(diǎn)是簡化了從 Kafka 遷移到 Pulsar 的流程，不需要更改兩遍代碼，直接無縫遷移。接下來就一起詳細(xì)看看KoP 的開發(fā)歷程吧。

什么是 Apache Pulsar

Apache Pulsar 是一個事件流平臺。最初，Apache Pulsar 就采用云原生、分層分片的架構(gòu)。該架構(gòu)將服務(wù)和存儲分離開來，使系統(tǒng)實(shí)現(xiàn)更友好的容器化。

而現(xiàn)在，Pulsar 不僅僅是是一個消息中間件，更是一個消息+流數(shù)據(jù)結(jié)合的系統(tǒng)，即 Cloud-Native Event Streaming。

我們之前寫過很多關(guān)于 Pulsar 的具體詳情，感興趣的可以查看 ：Apache Pulsar 介紹。

 Why KoP？

Plusar 為隊(duì)列和流工作負(fù)載提供統(tǒng)一的消息模型。Pulsar 支持自己基于 protobuf 的二進(jìn)制協(xié)議，以確保高性能和低延遲。protobuf 有利于實(shí)現(xiàn) Pulsar 客戶端。

而且，該項(xiàng)目也支持 Java，Go，Python 和 C ++ 語言以及社區(qū)提供的第三方客戶端。但是，對于使用其他消息傳輸協(xié)議編寫的應(yīng)用程序，用戶必須重寫這些應(yīng)用程序，否則這些應(yīng)用程序無法采用 Pulsar 新的統(tǒng)一消息傳輸協(xié)議。

為了解決這一問題，Pulsar 社區(qū)之前也開發(fā)了一些應(yīng)用程序，以便將 Kafka 應(yīng)用程序從其他消息系統(tǒng)遷移到 Pulsar。例如，Pulsar 在 Kafka Java API 上提供了 Kafka wrapper。

Kafka wrapper 允許用戶在不改變代碼的情況下將其使用的 Kafka Java 客戶端應(yīng)用程序從 Kafka 切換到 Pulsar。Pulsar 還提供豐富的 connector 生態(tài)系統(tǒng)，用于連接 Pulsar 和其他數(shù)據(jù)系統(tǒng)。

但是，那些想要從其他 Kafka 應(yīng)用程序切換到 Pulsar 的用戶仍然有強(qiáng)烈的需求。

KoP 的誕生背景

因此，就產(chǎn)生了“在 Pulsar 上支持 Kafka 協(xié)議”的想法。最初的猜想是添加一個 proxy，比如好多公司會在 Kafka 之前加一個類似 HTTP proxy，后續(xù)再轉(zhuǎn)換成 Pulsar 協(xié)議。

第二種猜想是能否直接將 Kafka 協(xié)議直接接入到 Pulsar broker 里，也就是目前 KoP 的成型。

那么關(guān)于第一種 proxy 做法，如果實(shí)現(xiàn)起來大概是什么樣呢？OVHcloud 就有過一次嘗試。

之前 OVHcloud 一直采用 Apache Kafka。盡管他們有在 Kafka 上運(yùn)行多個集群且每秒處理數(shù)百萬條消息的經(jīng)驗(yàn)，但仍面臨艱巨的運(yùn)營挑戰(zhàn)。所以，OVHcloud 放棄 Kafka，決定將其服務(wù)的產(chǎn)品轉(zhuǎn)移到 Pulsar，并在 Pulsar 上構(gòu)建其產(chǎn)品。

但是為了照顧到依舊使用 Kafka 系統(tǒng)的用戶，所以他們想在 Pulsar 里添加一個 proxy 去支持 Kafka 協(xié)議。他們最初的做法就是將 Kafka 協(xié)議的一幀轉(zhuǎn)換成 Pulsar 協(xié)議。

Proxy 收到來自 Kafka 客戶端的任何一幀，通過自由狀態(tài)機(jī)將其轉(zhuǎn)換為 Pulsar 相應(yīng)的接口。

這個狀態(tài)機(jī)一種是用于接收 Kafka 請求，第二種是用于處理 Pulsar response。然后在其中間再添加一個狀態(tài)機(jī)進(jìn)行同步。

因?yàn)樵?TCP 層進(jìn)行這些操作，所以它的表現(xiàn)還是不錯的。借由 Rust 的特性，整體運(yùn)行流暢。但是這個情況下，代碼仍需要一行行去寫，同時 Kafka 協(xié)議里有一些是沒有辦法通過 proxy 方式實(shí)現(xiàn)。比如：group coordinator 和 offsets management。

還有一個比較關(guān)鍵的點(diǎn)是，因?yàn)橛?Rust 去構(gòu)寫，所以比較難開源。即便是開源出來也很難作為一個組件去插入到 Pulsar 系統(tǒng)中。

剛好去年 StreamNative 的一條推特引起了 OVHcloud 的注意。這是 StreamNative 第一次舉行線下 Pulsar meetup 時翟佳老師分享的 KoP demo。

經(jīng)過幾次雙方經(jīng)驗(yàn)互談交流后，雙方合力推出了更完善的「KoP」。利用 Pulsar 和 BookKeeper 的事件流存儲架構(gòu)和 Pulsar 的可插拔協(xié)議處理插件框架來提供一種精簡而全面的解決方案。

KoP 組件與 Broker 協(xié)作

所以當(dāng)我們倒回去重看 Pulsar 架構(gòu)，下方模塊圖中最核心的：Broker、BookKeeper、ZooKeeper。Pulsar 就是基于 Managed ledger 實(shí)現(xiàn)的一套分布式流式存儲，包括如何存數(shù)據(jù)、如何防止數(shù)據(jù)丟失、流如何從本地機(jī)房復(fù)制到另一機(jī)房等。

Pulsar 協(xié)議本身是一個很輕量級的東西，即上圖中的 Pulsar protocol handler。它主要是處理 TCP 過來的請求格式，然后將請求轉(zhuǎn)化和讀取的操作。所以 Pulsar 協(xié)議最核心部分在存儲層面、分布式均衡層面等。

將 Pulsar protocol handler 抽象出來，變成一個框架/接口。利用這個框架，可以直接訪問 Pulsar 已經(jīng)構(gòu)建好的存儲系統(tǒng)，剩下要做的只是協(xié)議的解析和轉(zhuǎn)換。

所以依據(jù)這個構(gòu)想，將 Kafka 協(xié)議帶入去實(shí)踐。在  Pulsar 2.5 版本時新加了一個「Pluggable protocol handler」的概念（PIP-41），將接口單獨(dú)抽離了出來。

Pulsar protocol handler 的使用是類似 Pulsar function/connector，只需將其插入到 Pulsar broker 中，就可以讓 Pulsar 具有讀取和解析其他協(xié)議的能力。這個機(jī)制只需要調(diào)整兩個配置：

配置完成后，重啟集群即可支持「其他類型協(xié)議」的處理能力。當(dāng)然這個特性只在 Pulsar 2.5 版本后才支持，所以如需嘗試，可以先將 Pulsar 系統(tǒng)升級到 2.5 版本。

所以在此機(jī)制下過程就會變得更加明了簡單。只需在 Pulsar 里實(shí)現(xiàn) Kafka protocol handler 即可，剩下的上圖實(shí)線綠色部分是 Kafka 原生客戶端。只需將數(shù)據(jù)接入到 Pulsar 集群，就可以處理 Kafka 請求。

為什么選取 Kafka 作為實(shí)踐對象？

應(yīng)為 Pulsar 和 Kafka 在一些層面有很多相似之處。比如日志層，Pulsar 和 Kafka 都采用非常相似的數(shù)據(jù)模型，用于發(fā)布/訂閱消息和事件流，Pulsar 和 Kafka 都采用分布式日志。

通過對比 Pulsar 和 Kafka，我們發(fā)現(xiàn)這兩種系統(tǒng)有很多相似之處。這兩種系統(tǒng)都包括以下操作：

• Topic 查找
  所有客戶端都連接到任一 broker 以查找 Topic 的元數(shù)據(jù)（即 owner broker）。獲取元數(shù)據(jù)之后，客戶端與 owner broker 建立持久的 TCP 連接。
• 發(fā)布
  客戶端與 Topic 區(qū)的 owner broker 進(jìn)行對話，以將消息追加到分布式日志中。
• 消費(fèi)
  客戶端與 Topic 分區(qū)的 owner broker 進(jìn)行對話，以便從分布式日志中讀取消息。
• 偏移量
  為發(fā)布給 Topic 分區(qū)的消息分配偏移量。在 Pulsar 中，偏移量被稱為 MessageId。consumer 可以使用偏移量來查找日志中的給定位置，以便讀取消息。

• 消費(fèi)狀態(tài)
  這兩個系統(tǒng)都維護(hù)訂閱中的 consumer（ Kafka 稱之為消費(fèi)組）的消費(fèi)狀態(tài)。Kafka 將消費(fèi)狀態(tài)存儲在 `__offsets` Topic，而 Pulsar 將消費(fèi)狀態(tài)存儲在 `cursors`。

實(shí)現(xiàn)方式

1. Topic

Kafka 將所有 Topic 存儲在扁平的命名空間。但是，Pulsar 將 Topic 存儲在層次化、多租戶的命名空間。我們在 broker 配置中添加了 `kafkaNamespace` 配置，這樣管理員就可以將 Kafka Topic 映射到 Pulsar Topic。

為了方便 Kafka 用戶使用 Apache Pulsar 的多租戶特性，當(dāng) Kafka 用戶使用 SASL 驗(yàn)證機(jī)制來驗(yàn)證 Kafka 客戶端的時候，可以指定一個 Pulsar 租戶和命名空間作為其 SASL 用戶名。

2. 消息 ID 和偏移量

Kafka 為每條被成功發(fā)布到 Topic 分區(qū)的消息都指定了一個偏移量。Pulsar 為每條消息指定了一個 `MessageID`。消息 ID 由 `ledger-id`、 `entry-id` 和 `batch-index` 組成。我們在 Pulsar-Kafka wrapper 中使用相同的方法將 Pulsar 的消息 ID 轉(zhuǎn)換為偏移量，反之亦然。

3. 消息

Kafka 和 Pulsar 的消息都包含鍵、值、時間戳和 header（在 Pulsar 中被稱作 ‘properties’）。我們自動在 Kafka 消息和 Pulsar 消息之間轉(zhuǎn)換這些字段。

4. Topic 查找

我們?yōu)?Kafka 和 Pulsar 的請求處理插件提供相同的 Topic 查找方法。請求處理插件發(fā)現(xiàn) Topic，查找所請求的 Topic 分區(qū)的全部所有權(quán)，然后將包含所有權(quán)信息的 Kafka `TopicMetadata` 返回給 Kafka 客戶端。

5. 發(fā)布消息

當(dāng)收到 Kafka 客戶端發(fā)布的消息后，Kafka 請求處理插件逐一將多個字段（例如鍵、值、時間戳和 headers）進(jìn)行映射，從而將 Kafka 消息轉(zhuǎn)換為 Pulsar 消息。

同時，Kafka 請求處理插件利用 ManagedLedger append API 將這些已轉(zhuǎn)化的 Pulsar 消息存儲在 BookKeeper。Kafka 請求處理插件將 Kafka 消息轉(zhuǎn)換為 Pulsar 消息后，現(xiàn)有的 Pulsar 應(yīng)用程序就可以接收 Kafka 客戶端發(fā)布的消息。

6. 消費(fèi)消息

當(dāng)收到 Kafka 客戶端的 consumer 請求時，Kafka 請求處理插件打開一個非持久 cursor，然后從請求的偏移量開始讀取 entries。

Kafka 請求處理插件將 Pulsar 消息轉(zhuǎn)換回 Kafka 消息后，現(xiàn)有的 Kafka 應(yīng)用程序就可以接收 Pulsar 客戶端發(fā)布的消息。

7. Group coordinator & 偏移量管理

最大的挑戰(zhàn)是實(shí)現(xiàn) group coordinator 和偏移量管理。Pulsar 不支持集中的 group coordinator，無法為消費(fèi)組里的 consumer 分配分區(qū)，也無法管理每個消費(fèi)組的偏移量。

Pulsar broker 基于分區(qū)來管理分區(qū)分配，而分區(qū)的 owner broker 通過將確認(rèn)信息存儲在 cursors 來管理偏移量。

我們很難讓 Pulsar 模型與 Kafka 模型保持一致。因此，為了完全兼容 Kafka 客戶端，我們將 coordinator group 的更改和偏移量存儲在 Pulsar 名為    `public/kafka/__offsets`   系統(tǒng) Topic 中，從而實(shí)現(xiàn) Kafka coordinator group。

這樣，我們能夠在 Pulsar 和 Kafka 之間建立橋梁，并允許用戶使用現(xiàn)有的 Pulsar 工具和策略來管理訂閱并監(jiān)控 Kafka consumer。我們在已實(shí)現(xiàn)的 coordinator group 中添加一個后臺線程，定期將偏移量更新從系統(tǒng) Topic 同步到 Pulsar cursor。

因此，實(shí)際上 Kafka 消費(fèi)組被認(rèn)為是 Pulsar 訂閱。所有現(xiàn)有的 Pulsar 工具也可以用于管理 Kafka 消費(fèi)組。

KoP 生產(chǎn)化

如果將 KoP 應(yīng)用到實(shí)際場景中，就需要考慮以下多個方面：

• 多租戶
• 安全性
• 跨機(jī)房復(fù)制
• 分層存儲
• Schema
• 與已有的數(shù)據(jù)環(huán)境（如 Flink、Spark、Presto）集成

Q & A

1. Pulsar 有多種擴(kuò)展，這些擴(kuò)展有統(tǒng)一的管理方式嗎？

目前在做一個項(xiàng)目：Pulsar Registry，類似于 DocHub。也可以看作一個應(yīng)用商店，會集中一些組件/插件合集，可以期待一下。

2. Kafka 0.11 以下的版本是否能平滑升級到高版本？如果消息格式變了，是不是沒法平滑升級？

不能，0.10/0.11 版本以上才可以平滑升級。

KoP 最終的目的，是方便用戶將 Kafka 上已有的應(yīng)用遷移到 Pulsar 上，同時通過 KoP 的方式讓用戶可以更方便地構(gòu)建產(chǎn)品。未來 KoP 也會加大對 schema 和 Kafka 版本的支持與多兼容性。

上述就是小編為大家分享的Kafka-on-Pulsar 的開發(fā)歷程是怎樣的了，如果剛好有類似的疑惑，不妨參照上述分析進(jìn)行理解。如果想知道更多相關(guān)知識，歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道。

分享標(biāo)題：Kafka-on-Pulsar的開發(fā)歷程是怎樣的
當(dāng)前URL：http://www.muchs.cn/article18/ihehgp.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供用戶體驗(yàn)、電子商務(wù)、搜索引擎優(yōu)化、標(biāo)簽優(yōu)化、做網(wǎng)站、網(wǎng)站改版

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)