kafka-python

假设kafka主题中的消息是操作系统中mdb文件(MS Access)的归档路径(类似于:path/to/folder/archive_name。zip)。消费者获取存档路径，将它们解压缩到临时文件，然后将解压缩文件转换为临时CSV，然后加载到熊猫数据...

使用kafka Python时出错(kafka-Python 1。4。7) Kafka消费者抛出错误 kafka。errors。UnsupportedCodecError: UnsupportedCodecError:没有找到lz4压缩编解码器的库 ###在我阅读文档的时候找到了解决方案 h...

我需要开发一个Python程序，它将作为一个Kafka消费者，并在此基础上做一些处理。我使用kafka-Python，它在本地测试中做得很好。 但是，我的生产环境(RHEL7)需要Kerberos身份验证。Keytab文件将用于此目的。 我找...

我们有一个要求，我们需要故意不平衡的卡夫卡集群分配流量分布等3代理集群代理代理2 1 15% 5% 80%代理3和主题的消息发送到代理根据代理流量分布。 为了在使用kafka-Python的Python编程中实现这个逻辑，我们在main...

我使用下面的脚本从kafka -读取数据 from kafka import KafkaConsumer from json import loads import json consumer = KafkaConsumer('topicname',bootstrap_servers=['brokerip'],enable_auto_commit=True) ...

如何在aiokafka中添加json消息的模式？Kafka Connect不能没有它。 import asyncio import json import random import aiokafka from faker import Faker def serializer(value): return json.dumps(val...

我安装了kafka & kafka - Python使用Python3。6解释器。 导入错误已经消失了，但当我们尝试运行我的代码时，我仍然得到了这个消息: from kafka.admin import KafkaAdminClient, NewTopic 模块化模块:没有名...

我只需要监控我的Kafka集群是否启动。偶尔运行Kafka的机器会被关闭。如果群集不可用，我想发送电子邮件警告。 我可以创建生产者和消费者定期发送和接收假消息。有更简单的方法吗？ ###你可以使用https://github...

提问 我尝试使用Kafka - Python来访问Docker容器中的Kafka。我试图连接到Kafka的dockerized应用程序在同一网络的另一个容器中。当我尝试初始化KafkaAdminClient对象时出现错误: self._kafka_admin = KafkaAdminC...

我对如何在pycharm中创建kafka生产者和消费者感到困惑。我创建了一个produce。py from time import sleep from json import dumps from kafka import KafkaProducer producer = KafkaProducer( value_seriali...

我想使用Python创建一个新的kafka主题，当我尝试使用server="kafka:9092"创建一个KafkaAdminClient时，我得到了一个错误: self._kafka_admin = KafkaAdminClient( bootstrap_servers=[...

我试图提取消息的kafka队列基于特定的时间范围，说从10月2日下午1点到10月2日下午3点。我如何显式地定义这个时间范围并从这些时间范围中提取数据？ bootstrap_servers = ['servername.com'] topicName = 'sales_...

我有一个读取s3文件并将其写入主题的Lambda，这工作得很好。 然而，我需要从文件中读取凭证，因为这是Python Kafka生产者(https://kafka-Python。readthedocs。io/en/master/apidoc/KafkaProducer。html#kafka。K...

我想要从一个系统到另一个系统的实时流。 我使用的是kafka-Python，并且能够在本地直播流。 发现连接器将处理多个设备。有没有人能给我推荐一种在Python中使用连接器实现它的方法？ Kafka Connect是一个Java框...