python 消费 kafka 数据教程

下面我将为您讲解“Python消费Kafka数据教程”的完整攻略。

1. 安装依赖

在Python中消费Kafka数据需要使用kafka-python库，所以我们需要先安装该依赖，可以通过以下命令安装：

pip install kafka-python

2. 编写消费者代码

首先，我们需要指定Kafka集群的IP及端口，以及指定要消费的topic名称。示例代码如下：

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'my_topic',
    bootstrap_servers=['kafka-broker1:9092','kafka-broker2:9092','kafka-broker3:9092']
)

然后，我们可以使用for循环遍历消费者消息队列中的消息并对其进行处理，示例代码如下：

for msg in consumer:
    print(msg.value.decode('utf-8'))

3. 示例

下面，我将举两个简单的示例来说明如何基于Python消费Kafka数据。

示例一： 监听特定的Topic并将消息输出到文件

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'my_topic',
    bootstrap_servers=['kafka-broker1:9092','kafka-broker2:9092','kafka-broker3:9092']
)

with open('kafka_msgs.txt', 'w') as f:
    for msg in consumer:
        f.write(f"{msg.value.decode('utf-8')}\n")

代码中，我们通过with open语句打开文件，并通过for循环不断遍历消费者队列中的消息并将其写入到文件中。

示例二：将消息处理后写入MySQL数据库

import json
import mysql.connector
from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'my_topic',
    bootstrap_servers=['kafka-broker1:9092','kafka-broker2:9092','kafka-broker3:9092']
)

db = mysql.connector.connect(
    host='localhost',
    user='user',
    password='password',
    database='testdb'
)

cursor = db.cursor()

for msg in consumer:
    data = json.loads(msg.value.decode('utf-8'))
    if data['type'] == 'order':
        # 该操作仅为示例，需要自行根据实际情况编写代码
        cursor.execute(f"INSERT INTO orders (order_id, product_id, user_id) VALUES ('{data['order_id']}', '{data['product_id']}', '{data['user_id']}')")
        db.commit()

代码中，我们通过json模块解析消息的内容并对其进行处理，然后将处理后的数据插入到MySQL数据库中。

总结

通过本文的介绍，我们了解到了Python消费Kafka数据的完整攻略。在实际使用中，我们可以根据具体需求对示例代码进行修改和扩展。