4. Python连接Kafka：安装kafka-python库，编写第一个Producer发送模拟因子数据，Consumer消费数据

在完成了Kafka集群的搭建与主题创建后，本章将进入实战编码环节。我们将使用Python生态中最流行的Kafka客户端库——kafka-python，实现因子数据的生产与消费。这是流式因子漂移检测管道的起点与终点。

4.1 环境准备与库安装

确保Python版本 ≥ 3.8，并已激活虚拟环境。使用pip安装kafka-python库：

pip install kafka-python==2.0.2

验证安装：

python -c "import kafka; print(kafka.__version__)"

输出应为 2.0.2。

4.2 编写第一个Producer：发送模拟因子数据

因子数据通常为结构化数值，例如动量因子、波动率因子、价值因子等。我们模拟一个包含时间戳、因子名称、因子值、股票代码的JSON消息。

4.2.1 生产者核心参数

4.2.2 生产者代码实现

import json
import time
import random
from kafka import KafkaProducer

# 模拟因子数据生成器
def generate_factor_data():
    factors = ['momentum', 'volatility', 'value', 'size', 'quality']
    stocks = ['000001.SZ', '600519.SH', '300750.SZ', '601318.SH', '000858.SZ']
    return {
        'timestamp': int(time.time() * 1000),
        'stock_code': random.choice(stocks),
        'factor_name': random.choice(factors),
        'factor_value': round(random.uniform(-2.0, 2.0), 4),
        'version': 1
    }

# 创建生产者实例
producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    key_serializer=lambda k: k.encode('utf-8') if k else None,
    acks=1,
    retries=3
)

# 发送10条模拟因子数据
topic = 'factor-input'
for i in range(10):
    data = generate_factor_data()
    future = producer.send(
        topic,
        key=data['stock_code'],
        value=data
    )
    record_metadata = future.get(timeout=10)
    print(f"Sent: {data['factor_name']} = {data['factor_value']} | "
          f"Partition: {record_metadata.partition} | "
          f"Offset: {record_metadata.offset}")
    time.sleep(0.5)

producer.flush()
producer.close()
print("Producer finished.")

4.3 编写第一个Consumer：消费因子数据

Consumer从指定主题拉取消息，反序列化后打印或存储。我们实现一个简单的控制台消费者，验证数据流是否完整。

4.3.1 消费者核心参数

4.3.2 消费者代码实现

import json
from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'factor-input',
    bootstrap_servers='localhost:9092',
    group_id='factor-drift-group',
    value_deserializer=lambda v: json.loads(v.decode('utf-8')),
    key_deserializer=lambda k: k.decode('utf-8') if k else None,
    auto_offset_reset='earliest',
    enable_auto_commit=True,
    auto_commit_interval_ms=5000
)

print("Consumer started. Waiting for messages...")
try:
    for message in consumer:
        print(f"Received: Key={message.key} | "
              f"Partition={message.partition} | "
              f"Offset={message.offset} | "
              f"Value={message.value}")
except KeyboardInterrupt:
    print("Consumer stopped.")
finally:
    consumer.close()

4.4 生产者-消费者数据流架构

下图展示了本章实现的端到端数据流：

4.5 运行与验证

1. 启动Kafka集群（如已启动可跳过）。
2. 创建主题（若未创建）：kafka-topics.sh --create --topic factor-input --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1
3. 先启动Consumer（在一个终端）：python consumer.py
4. 再启动Producer（在另一个终端）：python producer.py
5. 观察输出：Consumer终端应逐条打印10条因子数据，包含分区和偏移量信息。

4.6 常见问题与避坑