如何在 Windows 上使用 Kafka-Python 进行实时流处理

Apache Kafka 是一个分布式流平台，可用于处理实时数据。Kafka 经常与 Python 结合使用，以构建实时流应用程序。

在本文中，我们将了解如何在 Windows 上开始使用 Kafka-Python 进行实时流处理。我们将讨论以下主题：

在 Windows 上安装 Kafka
安装 Kafka-Python
创建主题和生成消息
消费消息
Kafka 使用案例

在 Windows 上安装 Kafka

第一步是在 Windows 上安装 Kafka。您可以从 Apache Kafka 网站下载适用于 Windows 的 Kafka 二进制文件： https: //kafka.apache.org/downloads。

下载 Kafka 二进制文件后，将它们解压到您选择的目录中。Kafka 二进制文件包含许多用于管理 Kafka 的命令行工具。下面定义的步骤将有助于配置 Zookeeper 和 Kafka 服务器配置并在localhost上启动它。

1) 安装 Java JDK 8 或更高版本。
2) 从官方网站下载 Kafka 二进制文件，并将下载的文件解压缩到要保存 Kafka 文件的目录中。
3) 打开解压缩的 Kafka 文件内的配置文件夹，编辑 zookeeper.properties 文件。
4) 复制字段 dataDir 的路径，并在路径中添加 /zookeeper-data
5) 打开另一个命令提示符，使用以下命令启动 Zookeeper： .\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties
6) 打开另一个命令提示符，使用以下命令启动 Kafka： .\bin\windows\kafka-server-start.bat .\config\server.properties

安装 Kafka-Python

下一步是安装 Kafka-Python。Kafka-Python 是一个 Python 库，可以轻松地与 Kafka 进行交互。您可以使用 pip 安装 KafkaPython：

pip install kafka-python
or
conda install -c conda-forge kafka-python

创建主题并生成简单的消息

#KafkaProducer.py

from kafka import KafkaProducer
from kafka.errors import NoBrokersAvailable
import time

# Create a Kafka Producer on Localhost with Default Port
producer = KafkaProducer(bootstrap_servers='localhost:9092')

# Create a New Topic to Send Messages
topic = 'Test_Topic_1'

# Retry logic to handle connection issues
max_retries = 3
current_retry = 0
while current_retry < max_retries:
    try:
        for i in range(5):
            message = f'Message {i}'
            producer.send(topic, message.encode('utf-8'))
            print(f'Sent: {message}')
        # If the messages are sent successfully, break the loop
        break

    except NoBrokersAvailable as e:
        print(f"Error: {e}")
        print("Retrying in 5 seconds...")
        time.sleep(5)
        current_retry += 1

# Close the Producer and Close the connections
producer.close()

消费消息

#KafkaConsumer.py

from kafka import KafkaConsumer

# Create a Kafka Consumer
consumer = KafkaConsumer(
    'Test_Topic_1', # Topic Name
    bootstrap_servers='localhost:9092', #Host and Port
    group_id='your_consumer_group' # Group ID
)

# Continuously listen / poll for New Messages
for message in consumer:
    print(f'Received: {message.value.decode("utf-8")}')

Kafka 用于 JSON 数据

#Kafka_JSON_Producer.py

from kafka import KafkaProducer
import json

# Create a Kafka producer
producer = KafkaProducer(bootstrap_servers='localhost:9092', 
                         value_serializer=lambda m: json.dumps(m).encode('utf-8'))

# Create a JSON object
data = {'productname': 'Pen', 'price': 20, 'city': 'Chennai'}

# Send a JSON object via Kafka
producer.send('Test_Topic_1', value=data)

producer.close()