數(shù)據(jù)賦予了世界力量。 我們每秒獲取大量信息，我們對(duì)其進(jìn)行清理，分析并創(chuàng)建更有價(jià)值的輸出，無論是日志文件，用戶活動(dòng)，聊天消息還是其他內(nèi)容。 我們提供的速度越快，便會(huì)為客戶帶來更多的價(jià)值。 我們正處于一個(gè)快節(jié)奏且瞬息萬變的環(huán)境時(shí)代。

[[328122]]

Apache Kafka是一個(gè)分布式流平臺(tái)，可以實(shí)時(shí)發(fā)布，訂閱，存儲(chǔ)和處理消息。 其基于拉式的體系結(jié)構(gòu)減輕了重負(fù)載對(duì)服務(wù)的壓力，使其易于擴(kuò)展。 它以低延遲將大量數(shù)據(jù)從源移動(dòng)到目的地。

關(guān)于推與拉架構(gòu)的思考

我最近與人們討論了不同服務(wù)架構(gòu)的優(yōu)缺點(diǎn)…

Kafka是基于JVM的平臺(tái)，因此客戶端的主流編程語言是Java。 但是，隨著社區(qū)的蓬勃發(fā)展，高質(zhì)量的開源Python客戶端也已面世，并已用于生產(chǎn)中。

在本文中，我將介紹最著名的Python Kafka客戶端：kafka-python，pykafka和confluent-kafka并進(jìn)行比較。 最后，我將對(duì)每個(gè)圖書館的利弊發(fā)表自己的看法。

我們?yōu)槭裁匆狵afka?

首先是第一件事。 為什么選擇Kafka? Kafka旨在增強(qiáng)事件驅(qū)動(dòng)的體系結(jié)構(gòu)。 它通過提供高吞吐量，低延遲，高耐用性和高可用性解決方案來增強(qiáng)體系結(jié)構(gòu)。 (這并不意味著您可以同時(shí)擁有所有它們，總會(huì)有一個(gè)權(quán)衡。閱讀此白皮書以了解更多信息。)

如何為高性能和低延遲部署和優(yōu)化Kafka

Apache Kafka®是一個(gè)功率流處理平臺(tái)，他的白皮書討論了如何針對(duì)以下情況優(yōu)化Kafka部署：

除了其高性能外，另一個(gè)吸引人的功能是發(fā)布/訂閱模型，其中發(fā)件人沒有專門向收件人發(fā)送郵件。 而是根據(jù)主題將郵件傳遞到收件人可以訂閱的集中位置。

這樣，我們可以輕松地將應(yīng)用程序解耦并擺脫整體設(shè)計(jì)。 讓我們看一個(gè)例子，了解為什么去耦效果更好。

您創(chuàng)建的網(wǎng)站需要將用戶活動(dòng)發(fā)送到某個(gè)地方，因此您可以編寫從網(wǎng)站到實(shí)時(shí)監(jiān)控儀表板的直接連接。 這是一個(gè)簡(jiǎn)單的解決方案，效果很好。 有一天，您決定將用戶活動(dòng)存儲(chǔ)在數(shù)據(jù)庫中以備將來分析。 因此，您將另一個(gè)直接數(shù)據(jù)庫連接寫入到您的網(wǎng)站。 同時(shí)，您的網(wǎng)站越來越多的流量，并且您想通過添加警報(bào)服務(wù)，實(shí)時(shí)分析服務(wù)等來增強(qiáng)它的功能。

您的架構(gòu)最終將像這樣。 諸如大量代碼倉庫，安全性問題，可伸縮性問題和可維護(hù)性問題之類的問題將傷害您。

> Architecture without decoupling (Created by Xiaoxu Gao)

您需要一個(gè)中心來分隔具有不同角色的應(yīng)用程序。 對(duì)于創(chuàng)建事件的應(yīng)用程序，我們稱它們?yōu)樯a(chǎn)者。 他們將事件發(fā)布到集中式中心。 每個(gè)事件(即消息)都屬于一個(gè)主題。 消費(fèi)者位于樞紐的另一側(cè)。 他們從中心訂閱了他們需要的主題，而無需直接與制作人交談。

有了此模型，就可以輕松擴(kuò)展和維護(hù)體系結(jié)構(gòu)。 工程師可以將更多精力放在核心業(yè)務(wù)上。

> Architecture with decoupling (Created by Xiaoxu Gao)

簡(jiǎn)而言之，Kafka設(shè)置

您可以從官方網(wǎng)站下載Apache Kafka。 快速入門可幫助您在10秒鐘內(nèi)啟動(dòng)服務(wù)器。

您也可以從Confluent平臺(tái)下載Apache Kafka。 它是迄今為止最大的面向Kafka的流數(shù)據(jù)平臺(tái)。 它為個(gè)人和企業(yè)提供了一系列圍繞Kafka的基礎(chǔ)架構(gòu)服務(wù)，以便將數(shù)據(jù)作為實(shí)時(shí)流提供。 創(chuàng)始人是最初創(chuàng)建Apache Kafka的團(tuán)隊(duì)的成員。

每臺(tái)Kafka服務(wù)器都稱為代理，您可以以獨(dú)立模式運(yùn)行它或形成集群。 除了Kafka，我們還需要Zookeeper存儲(chǔ)有關(guān)Kafka的元數(shù)據(jù)。 Zookeeper的行為就像協(xié)調(diào)器一樣，負(fù)責(zé)管理分布式系統(tǒng)中每個(gè)代理的狀態(tài)。

> Kafka setup (Created by Xiaoxu Gao)

假設(shè)我們已經(jīng)與1位Zookeeper和1位Kafka經(jīng)紀(jì)人建立了基礎(chǔ)架構(gòu)。 現(xiàn)在該連接了! 原始的Java客戶端提供5個(gè)API：

• 生產(chǎn)者API：將消息發(fā)布到Kafka集群中的主題。
• 使用者API：使用來自Kafka集群中主題的消息。
• Streams API：使用主題中的消息，并將其轉(zhuǎn)換為Kafka集群中的其他主題。 這些操作可以是過濾，聯(lián)接，映射，分組等。
• 連接API：無需編碼即可直接將Kafka群集連接到源系統(tǒng)或接收器系統(tǒng)。 該系統(tǒng)可以是文件，關(guān)系數(shù)據(jù)庫，Elasticsearch等。
• 管理員API：管理和檢查Kafka集群中的主題和代理。

Kafka的Python庫

在Python世界中，已經(jīng)實(shí)現(xiàn)了5個(gè)API中的3個(gè)，分別是Producer API，Consumer API和Admin API。 Python中還沒有這樣的Kafka Stream API，但是很好的替代方法是Faust。

本節(jié)中的測(cè)試是基于本地安裝的1個(gè)Zookeeper和1個(gè)Kafka代理執(zhí)行的。 這與性能調(diào)整無關(guān)，所以我主要使用該庫提供的默認(rèn)配置。

Kafka-Python

kafka-python的設(shè)計(jì)功能非常類似于官方的Java客戶端，并帶有大量pythonic接口。 最好與Kafka 0.9+版本一起使用。 第一版發(fā)布于2014年3月。正在積極維護(hù)中。

安裝

pip install kafka-python

每個(gè)消息都是通過send()異步發(fā)送的。 調(diào)用時(shí)，它將記錄添加到緩沖區(qū)并立即返回。 這使生產(chǎn)者可以以批處理方式將記錄發(fā)送到Kafka經(jīng)紀(jì)人以提高效率。 異步可以極大地提高速度，但是我們還應(yīng)該了解以下幾點(diǎn)：

• 在異步模式下，不能保證排序。 您無法控制Kafka經(jīng)紀(jì)人何時(shí)確認(rèn)(確認(rèn))每封郵件。
• 為生產(chǎn)者提供成功回調(diào)和失敗回調(diào)是一個(gè)好習(xí)慣。 例如，您可以在成功回調(diào)中編寫信息日志消息，而在失敗回調(diào)中編寫異常日志消息。
• 由于無法保證順序，因此在回調(diào)中收到異常之前，可能會(huì)發(fā)送額外的消息。

如果要避免這些問題，可以選擇同步發(fā)送消息。 send()的返回是FutureRecordMetadata。 通過執(zhí)行future.get(timeout = 60)，生產(chǎn)者將被阻止最多60秒鐘，直到代理成功確認(rèn)消息為止。 缺點(diǎn)是速度，與異步模式相比，它相對(duì)較慢。

消費(fèi)者

使用者實(shí)例是一個(gè)Python迭代器。 消費(fèi)者類的核心是poll()方法。 它允許使用者繼續(xù)從主題中提取消息。 它的輸入?yún)?shù)timeout_ms之一默認(rèn)為0，這意味著該方法將立即返回所有在緩沖區(qū)中拉出并可用的記錄。 您可以增加timeout_ms以返回更大的批次。

默認(rèn)情況下，每個(gè)使用者都是一個(gè)無限的偵聽器，因此它不會(huì)停止運(yùn)行，直到程序中斷。 但另一方面，您可以根據(jù)收到的消息停止使用者。 例如，您可以退出循環(huán)并在達(dá)到某個(gè)偏移量時(shí)關(guān)閉使用者。

也可以將使用者分配給一個(gè)分區(qū)或來自多個(gè)主題的多個(gè)分區(qū)。

這是kafka-python庫的測(cè)試結(jié)果。 每個(gè)消息的大小為100字節(jié)。 生產(chǎn)者的平均吞吐量為1.4MB / s。 使用者的平均吞吐量為2.8MB / s。

Confluent-kafka

Confluent-kafka是Python的高性能Kafka客戶端，它利用高性能C客戶端librdkafka。 從1.0版開始，這些作為PyPi上的OS X和Linux的獨(dú)立二進(jìn)制輪分發(fā)。 它支持Kafka 0.8+版本。 第一版發(fā)布于2016年5月。正在積極維護(hù)中。

安裝

對(duì)于OS X和Linux，軟件包中包括librdkafka，需要單獨(dú)安裝。

pip install confluent-kafka

對(duì)于Windows用戶，在我撰寫本文時(shí)，confluent-kafka尚未在Windows上支持Python3.8二進(jìn)制輪子。 您將遇到librdkafka的問題。 請(qǐng)查看他們的發(fā)行說明，該說明正在積極開發(fā)中。 另一種解決方案是降級(jí)到Python3.7。

Confluent-kafka在速度方面具有令人難以置信的性能。 API的設(shè)計(jì)有點(diǎn)類似于kafka-python。 您可以通過將flush()放入循環(huán)中來使其同步。

消費(fèi)者

confluent-kafka中的Consumer API需要更多代碼。 您無需自己處理高級(jí)循環(huán)方法(例如，消耗())，而需要自己處理while循環(huán)。 我建議您創(chuàng)建自己的consump()，它實(shí)際上是一個(gè)Python生成器。 只要有一條消息被拉出并且在緩沖區(qū)中可用，它就會(huì)產(chǎn)生該消息。

這樣，主要功能將變得干凈，您可以自由控制消費(fèi)者的行為。 例如，您可以在consumpt()中定義一個(gè)"會(huì)話窗口"。 如果在X秒鐘內(nèi)未提取任何消息，則使用者將停止。 或者，您可以添加標(biāo)志infinite = True作為輸入?yún)?shù)，以控制使用者是否應(yīng)為無限偵聽器。

這是confluent-kafka庫的測(cè)試結(jié)果。 每個(gè)消息的大小為100字節(jié)。 生產(chǎn)者的平均吞吐量為21.97MBps。 消費(fèi)者的平均吞吐量為16.8〜28.7MB / s。

PyKafka

PyKafka是Python的程序員友好的Kafka客戶端。 它包括Kafka生產(chǎn)者和使用者的Python實(shí)現(xiàn)，可以選擇由基于librdkafka的C擴(kuò)展支持。 它支持Kafka 0.82+版本。 第一版發(fā)布于2012年8月，但自2018年11月以來未進(jìn)行過更新。

安裝

pip install pykafka

該軟件包不附帶librdkafka，您需要在所有操作系統(tǒng)中分別安裝。

pykafka具有KafkaClient接口，該接口涵蓋了ProducerAPI和Consumer API。

消息可以異步和同步模式發(fā)送。 我發(fā)現(xiàn)pykafka會(huì)修改某些生產(chǎn)者配置(例如linger_ms和min_queued_messages)的默認(rèn)值，這會(huì)對(duì)發(fā)送少量數(shù)據(jù)產(chǎn)生影響。

您可以將其與Apache Kafka網(wǎng)站上的默認(rèn)配置進(jìn)行比較。

如果要獲取每個(gè)消息的回調(diào)，請(qǐng)確保將min_queued_messages更改為1，否則如果數(shù)據(jù)集小于70000，則不會(huì)收到任何報(bào)告。

> pykafka-producer-config

消費(fèi)者

您可以從KafkaClinet界面獲取SimpleConsumer。 這類似于kafka-python，其中民意調(diào)查被包裝在SimpleConsumer類中。

這是pykafka庫的測(cè)試結(jié)果。 每個(gè)消息的大小為100字節(jié)。 生產(chǎn)者的平均吞吐量為2.1MB / s。 使用者的平均吞吐量為1.57MB / s。

結(jié)論

到目前為止，我已經(jīng)解釋了每個(gè)庫的Producer API和Consumer API。 就Admin API而言，kafka-python和confluent-kafka確實(shí)提供了顯式的Admin API。 您可以在要?jiǎng)?chuàng)建主題的單元測(cè)試中使用它，然后在執(zhí)行下一個(gè)測(cè)試之前將其刪除。 此外，如果您想使用Python構(gòu)建Kafka監(jiān)控儀表板，則Admin API可以幫助您檢索集群和主題的元數(shù)據(jù)。

Confluent-kafka：

毫無疑問，Confluent-kafka在這三個(gè)庫中表現(xiàn)最佳。 該API的設(shè)計(jì)經(jīng)過精心設(shè)計(jì)，參數(shù)與原始Apache Kafka相同的名稱和默認(rèn)值。 您可以輕松地將其鏈接到原始參數(shù)。 就個(gè)人而言，我喜歡自定義消費(fèi)者行為的靈活性。 Confluent也正在積極開發(fā)和支持它。

缺點(diǎn)是Windows用戶可能需要花費(fèi)一些時(shí)間才能使其工作。 并且由于C擴(kuò)展，調(diào)試可能很棘手。

kafka-python：

kafka-python是沒有C擴(kuò)展的純Python庫。 該API經(jīng)過精心設(shè)計(jì)，對(duì)于初學(xué)者來說很容易使用。 這也是一個(gè)積極開發(fā)的項(xiàng)目。

python-kafka的缺點(diǎn)是它的速度。 如果您確實(shí)關(guān)心性能，建議您改用confluent-kafka。

pykafka：

與kafka-python和conflunet-kafka相比，pykafka的開發(fā)活動(dòng)較少。 該版本的歷史記錄表明，自2018年11月以來尚未進(jìn)行過更新。此外，pykafka具有不同的API設(shè)計(jì)，并使用了不同的默認(rèn)參數(shù)，這可能不是第一次。