2. 环境准备：Python环境配置、Kafka与Zookeeper的安装与启动、依赖库安装

在开始构建因子漂移检测系统之前，必须搭建一个稳定、可复现的实验环境。本章将指导你完成从基础运行环境到核心流处理组件的全链路配置，确保后续章节的代码能够无缝运行。

2.1 环境全景概览

本课程的技术栈围绕 Python 3.9+、Apache Kafka 3.5+ 和 Zookeeper 3.8+ 构建。下图展示了各组件之间的依赖关系与数据流向：

2.2 Python 环境配置

推荐使用 Anaconda 或 venv 创建独立的虚拟环境，避免包版本冲突。

2.2.1 创建虚拟环境

# 使用 conda 创建（推荐）
conda create -n drift_detection python=3.9 -y
conda activate drift_detection

# 或使用 venv
python3.9 -m venv drift_env
source drift_env/bin/activate  # Linux/Mac
# drift_env\Scripts\activate   # Windows

2.2.2 安装核心依赖库

本课程所需的 Python 库清单如下表所示：

库名称	版本要求	用途
kafka-python	≥ 2.0.2	Kafka 生产者/消费者客户端
scikit-learn	≥ 1.2.0	漂移检测算法（KS检验、PSI计算）
pandas	≥ 1.5.0	数据框操作与因子计算
numpy	≥ 1.23.0	数值计算与数组操作
matplotlib	≥ 3.6.0	漂移趋势可视化
joblib	≥ 1.2.0	模型序列化与缓存

# 一键安装所有依赖
pip install kafka-python==2.0.2 scikit-learn==1.2.2 pandas==1.5.3 numpy==1.23.5 matplotlib==3.6.3 joblib==1.2.0

2.3 Kafka 与 Zookeeper 的安装与启动

Kafka 依赖 Zookeeper 管理集群元数据。以下步骤基于 Kafka 3.5.1 和 Zookeeper 3.8.3，适用于 Linux/Mac 系统。Windows 用户可使用 WSL2 或下载二进制包后调整路径。

2.3.1 下载与解压

# 下载 Kafka（内置 Zookeeper）
wget https://downloads.apache.org/kafka/3.5.1/kafka_2.13-3.5.1.tgz
tar -xzf kafka_2.13-3.5.1.tgz
cd kafka_2.13-3.5.1

2.3.2 启动 Zookeeper

# 启动 Zookeeper（使用 Kafka 内置脚本）
bin/zookeeper-server-start.sh config/zookeeper.properties &

# 验证 Zookeeper 是否启动成功
# 默认监听 2181 端口
netstat -an | grep 2181

2.3.3 启动 Kafka Broker

# 修改配置文件（可选：调整数据保留时间等参数）
# vim config/server.properties

# 启动 Kafka Broker
bin/kafka-server-start.sh config/server.properties &

# 验证 Kafka 是否启动成功
# 默认监听 9092 端口
netstat -an | grep 9092

2.3.4 创建测试 Topic

# 创建用于因子漂移检测的 Topic
bin/kafka-topics.sh --create \
  --bootstrap-server localhost:9092 \
  --replication-factor 1 \
  --partitions 3 \
  --topic factor-stream

# 列出所有 Topic 确认创建成功
bin/kafka-topics.sh --list --bootstrap-server localhost:9092

2.4 验证环境连通性

启动 Kafka 后，使用 Python 快速验证生产者与消费者能否正常通信：

# test_connection.py
from kafka import KafkaProducer, KafkaConsumer
import json

# 生产者测试
producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)
producer.send('factor-stream', {'factor': 'test', 'value': 1.0})
producer.flush()
print("✅ 生产者发送成功")

# 消费者测试
consumer = KafkaConsumer(
    'factor-stream',
    bootstrap_servers='localhost:9092',
    auto_offset_reset='earliest',
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)
for msg in consumer:
    print(f"✅ 消费者接收: {msg.value}")
    break
consumer.close()

# 运行测试脚本
python test_connection.py

若控制台输出两条确认信息，则说明 Kafka 环境配置成功，可以进入下一章的数据流构建。

2.5 环境配置清单速查

下表汇总了所有需要检查的环境要素，建议逐项核对：

检查项	预期结果	验证命令
Python 版本	3.9.x	python --version
虚拟环境激活	drift_detection	conda info --envs
kafka-python 安装	2.0.2	pip show kafka-python
Zookeeper 进程	端口 2181 监听	netstat -an | grep 2181
Kafka Broker 进程	端口 9092 监听	netstat -an | grep 9092
Topic 创建	factor-stream 存在	kafka-topics.sh --list
Python 连通性	生产/消费成功	python test_connection.py