如何通过Python创建Pulsar流与Kafka流？

一、创建 Pulsar 流 #

使用官方 pulsar-client 库即可快速实现Pulsar流的生产与消费：

1. 安装依赖

pip install pulsar-client

2. 生产者示例（发送消息到Pulsar Topic）

import pulsar

# 连接Pulsar服务
client = pulsar.Client('pulsar://localhost:6650')
# 创建生产者绑定指定Topic
producer = client.create_producer('my-pulsar-topic')

# 批量发送消息
for idx in range(10):
    message = f"Pulsar stream message {idx}".encode('utf-8')
    producer.send(message)
    print(f"Sent message: {message.decode('utf-8')}")

# 关闭资源
producer.close()
client.close()

3. 消费者示例（从Pulsar Topic接收消息）

import pulsar

client = pulsar.Client('pulsar://localhost:6650')
# 创建消费者，指定Topic与订阅名称
consumer = client.subscribe('my-pulsar-topic', subscription_name='pulsar-sub-1')

try:
    while True:
        # 阻塞接收消息
        msg = consumer.receive()
        print(f"Received message: {msg.data().decode('utf-8')}")
        # 确认消息已处理，避免重复消费
        consumer.acknowledge(msg)
except KeyboardInterrupt:
    print("Consumer stopped manually")
finally:
    consumer.close()
    client.close()

二、使用 GQLAlchemy 创建 Kafka 流 #

GQLAlchemy 主要用于图数据库操作，通常结合Kafka实现流数据导入图数据库的场景，以下是具体实现：

1. 安装依赖

pip install gqlalchemy kafka-python

2. 流数据处理示例（Kafka生产数据 + GQLAlchemy写入图数据库）

from gqlalchemy import Memgraph
from kafka import KafkaProducer, KafkaConsumer
import json

# 连接Memgraph（GQLAlchemy默认适配的图数据库）
memgraph = Memgraph(host="localhost", port=7687)

# Kafka生产者：向Topic发送JSON格式数据
kafka_producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda data: json.dumps(data).encode('utf-8')
)
# 发送测试数据
sample_data = {"user_id": 1001, "action": "click", "timestamp": "2024-05-20T10:00:00"}
kafka_producer.send('user-behavior-topic', value=sample_data)
kafka_producer.flush()

# Kafka消费者：从Topic拉取数据，通过GQLAlchemy写入图数据库
kafka_consumer = KafkaConsumer(
    'user-behavior-topic',
    bootstrap_servers='localhost:9092',
    value_deserializer=lambda msg: json.loads(msg.decode('utf-8'))
)

for msg in kafka_consumer:
    record = msg.value
    # 执行Cypher语句创建图节点与关系
    memgraph.execute(
        f"""
        CREATE (u:User {{id: {record['user_id']}}})
        CREATE (a:Action {{type: "{record['action']}", time: "{record['timestamp']}"}})
        CREATE (u)-[:PERFORMED]->(a)
        """
    )
    print(f"Processed and saved record: {record}")

如果仅需单纯创建Kafka流（无需图数据库集成），直接使用 kafka-python 库即可，GQLAlchemy并非Kafka流处理的必需依赖。

内容的提问来源于stack exchange，提问作者KWriter