您好，登录后才能下订单哦！
密码登录
登录注册
点击 登录注册 即表示同意《亿速云用户服务条款》
本篇文章给大家分享的是有关使用Python掌握Apache Kafka应当了解的3个库分别是哪些，小编觉得挺实用的，因此分享给大家学习，希望大家阅读完这篇文章后可以有所收获，话不多说，跟着小编一起来看看吧。
Apache Kafka是一个分布式流平台，可以实时发布，订阅，存储和处理消息。 其基于拉式的体系结构减轻了重负载对服务的压力，使其易于扩展。 它以低延迟将大量数据从源移动到目的地。
Kafka是基于JVM的平台，因此客户端的主流编程语言是Java。 但是，随着社区的蓬勃发展，高质量的开源Python客户端也已面世，并已用于生产中。
在本文中，我将介绍最著名的Python Kafka客户端：kafka-python，pykafka和confluent-kafka并进行比较。 最后，我将对每个图书馆的利弊发表自己的看法。
首先是第一件事。 为什么选择Kafka? Kafka旨在增强事件驱动的体系结构。 它通过提供高吞吐量，低延迟，高耐用性和高可用性解决方案来增强体系结构。 (这并不意味着您可以同时拥有所有它们，总会有一个权衡。阅读此白皮书以了解更多信息。)
Apache Kafka®是一个功率流处理平台，他的白皮书讨论了如何针对以下情况优化Kafka部署：
除了其高性能外，另一个吸引人的功能是发布/订阅模型，其中发件人没有专门向收件人发送邮件。 而是根据主题将邮件传递到收件人可以订阅的集中位置。
这样，我们可以轻松地将应用程序解耦并摆脱整体设计。 让我们看一个例子，了解为什么去耦效果更好。
您创建的网站需要将用户活动发送到某个地方，因此您可以编写从网站到实时监控仪表板的直接连接。 这是一个简单的解决方案，效果很好。 有一天，您决定将用户活动存储在数据库中以备将来分析。 因此，您将另一个直接数据库连接写入到您的网站。 同时，您的网站越来越多的流量，并且您想通过添加警报服务，实时分析服务等来增强它的功能。
您的架构最终将像这样。 诸如大量代码仓库，安全性问题，可伸缩性问题和可维护性问题之类的问题将伤害您。
> Architecture without decoupling (Created by Xiaoxu Gao)
您需要一个中心来分隔具有不同角色的应用程序。 对于创建事件的应用程序，我们称它们为生产者。 他们将事件发布到集中式中心。 每个事件(即消息)都属于一个主题。 消费者位于枢纽的另一侧。 他们从中心订阅了他们需要的主题，而无需直接与制作人交谈。
有了此模型，就可以轻松扩展和维护体系结构。 工程师可以将更多精力放在核心业务上。
> Architecture with decoupling (Created by Xiaoxu Gao)
您可以从官方网站下载Apache Kafka。 快速入门可帮助您在10秒钟内启动服务器。
您也可以从Confluent平台下载Apache Kafka。 它是迄今为止最大的面向Kafka的流数据平台。 它为个人和企业提供了一系列围绕Kafka的基础架构服务，以便将数据作为实时流提供。 创始人是最初创建Apache Kafka的团队的成员。
每台Kafka服务器都称为代理，您可以以独立模式运行它或形成集群。 除了Kafka，我们还需要Zookeeper存储有关Kafka的元数据。 Zookeeper的行为就像协调器一样，负责管理分布式系统中每个代理的状态。
> Kafka setup (Created by Xiaoxu Gao)
假设我们已经与1位Zookeeper和1位Kafka经纪人建立了基础架构。 现在该连接了! 原始的Java客户端提供5个API：
生产者API：将消息发布到Kafka集群中的主题。
使用者API：使用来自Kafka集群中主题的消息。
Streams API：使用主题中的消息，并将其转换为Kafka集群中的其他主题。 这些操作可以是过滤，联接，映射，分组等。
连接API：无需编码即可直接将Kafka群集连接到源系统或接收器系统。 该系统可以是文件，关系数据库，Elasticsearch等。
管理员API：管理和检查Kafka集群中的主题和代理。
在Python世界中，已经实现了5个API中的3个，分别是Producer API，Consumer API和Admin API。 Python中还没有这样的Kafka Stream API，但是很好的替代方法是Faust。
本节中的测试是基于本地安装的1个Zookeeper和1个Kafka代理执行的。 这与性能调整无关，所以我主要使用该库提供的默认配置。
Kafka-Python
kafka-python的设计功能非常类似于官方的Java客户端，并带有大量pythonic接口。 最好与Kafka 0.9+版本一起使用。 第一版发布于2014年3月。正在积极维护中。
安装
pip install kafka-python
每个消息都是通过send()异步发送的。 调用时，它将记录添加到缓冲区并立即返回。 这使生产者可以以批处理方式将记录发送到Kafka经纪人以提高效率。 异步可以极大地提高速度，但是我们还应该了解以下几点：
在异步模式下，不能保证排序。 您无法控制Kafka经纪人何时确认(确认)每封邮件。
为生产者提供成功回调和失败回调是一个好习惯。 例如，您可以在成功回调中编写信息日志消息，而在失败回调中编写异常日志消息。
由于无法保证顺序，因此在回调中收到异常之前，可能会发送额外的消息。
如果要避免这些问题，可以选择同步发送消息。 send()的返回是FutureRecordMetadata。 通过执行future.get(timeout = 60)，生产者将被阻止最多60秒钟，直到代理成功确认消息为止。 缺点是速度，与异步模式相比，它相对较慢。
消费者
使用者实例是一个Python迭代器。 消费者类的核心是poll()方法。 它允许使用者继续从主题中提取消息。 它的输入参数timeout_ms之一默认为0，这意味着该方法将立即返回所有在缓冲区中拉出并可用的记录。 您可以增加timeout_ms以返回更大的批次。
默认情况下，每个使用者都是一个无限的侦听器，因此它不会停止运行，直到程序中断。 但另一方面，您可以根据收到的消息停止使用者。 例如，您可以退出循环并在达到某个偏移量时关闭使用者。
也可以将使用者分配给一个分区或来自多个主题的多个分区。
这是kafka-python库的测试结果。 每个消息的大小为100字节。 生产者的平均吞吐量为1.4MB / s。 使用者的平均吞吐量为2.8MB / s。
Confluent-kafka
Confluent-kafka是Python的高性能Kafka客户端，它利用高性能C客户端librdkafka。 从1.0版开始，这些作为PyPi上的OS X和Linux的独立二进制轮分发。 它支持Kafka 0.8+版本。 第一版发布于2016年5月。正在积极维护中。
安装
对于OS X和Linux，软件包中包括librdkafka，需要单独安装。
pip install confluent-kafka
对于Windows用户，在我撰写本文时，confluent-kafka尚未在Windows上支持Python3.8二进制轮子。 您将遇到librdkafka的问题。 请查看他们的发行说明，该说明正在积极开发中。 另一种解决方案是降级到Python3.7。
Confluent-kafka在速度方面具有令人难以置信的性能。 API的设计有点类似于kafka-python。 您可以通过将flush()放入循环中来使其同步。
消费者
confluent-kafka中的Consumer API需要更多代码。 您无需自己处理高级循环方法(例如，消耗())，而需要自己处理while循环。 我建议您创建自己的consump()，它实际上是一个Python生成器。 只要有一条消息被拉出并且在缓冲区中可用，它就会产生该消息。
这样，主要功能将变得干净，您可以自由控制消费者的行为。 例如，您可以在consumpt()中定义一个"会话窗口"。 如果在X秒钟内未提取任何消息，则使用者将停止。 或者，您可以添加标志infinite = True作为输入参数，以控制使用者是否应为无限侦听器。
这是confluent-kafka库的测试结果。 每个消息的大小为100字节。 生产者的平均吞吐量为21.97MBps。 消费者的平均吞吐量为16.8〜28.7MB / s。
PyKafka
PyKafka是Python的程序员友好的Kafka客户端。 它包括Kafka生产者和使用者的Python实现，可以选择由基于librdkafka的C扩展支持。 它支持Kafka 0.82+版本。 第一版发布于2012年8月，但自2018年11月以来未进行过更新。
安装
pip install pykafka
该软件包不附带librdkafka，您需要在所有操作系统中分别安装。
pykafka具有KafkaClient接口，该接口涵盖了ProducerAPI和Consumer API。
消息可以异步和同步模式发送。 我发现pykafka会修改某些生产者配置(例如linger_ms和min_queued_messages)的默认值，这会对发送少量数据产生影响。
您可以将其与Apache Kafka网站上的默认配置进行比较。
如果要获取每个消息的回调，请确保将min_queued_messages更改为1，否则如果数据集小于70000，则不会收到任何报告。
> pykafka-producer-config
消费者
您可以从KafkaClinet界面获取SimpleConsumer。 这类似于kafka-python，其中民意调查被包装在SimpleConsumer类中。
这是pykafka库的测试结果。 每个消息的大小为100字节。 生产者的平均吞吐量为2.1MB / s。 使用者的平均吞吐量为1.57MB / s。
到目前为止，我已经解释了每个库的Producer API和Consumer API。 就Admin API而言，kafka-python和confluent-kafka确实提供了显式的Admin API。 您可以在要创建主题的单元测试中使用它，然后在执行下一个测试之前将其删除。 此外，如果您想使用Python构建Kafka监控仪表板，则Admin API可以帮助您检索集群和主题的元数据。
Confluent-kafka：
毫无疑问，Confluent-kafka在这三个库中表现最佳。 该API的设计经过精心设计，参数与原始Apache Kafka相同的名称和默认值。 您可以轻松地将其链接到原始参数。 就个人而言，我喜欢自定义消费者行为的灵活性。 Confluent也正在积极开发和支持它。
缺点是Windows用户可能需要花费一些时间才能使其工作。 并且由于C扩展，调试可能很棘手。
kafka-python：
kafka-python是没有C扩展的纯Python库。 该API经过精心设计，对于初学者来说很容易使用。 这也是一个积极开发的项目。
python-kafka的缺点是它的速度。 如果您确实关心性能，建议您改用confluent-kafka。
pykafka：
与kafka-python和conflunet-kafka相比，pykafka的开发活动较少。 该版本的历史记录表明，自2018年11月以来尚未进行过更新。此外，pykafka具有不同的API设计，并使用了不同的默认参数，这可能不是第一次。
以上就是使用Python掌握Apache Kafka应当了解的3个库分别是哪些，小编相信有部分知识点可能是我们日常工作会见到或用到的。希望你能通过这篇文章学到更多知识。更多详情敬请关注亿速云行业资讯频道。
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。