第4章环境搭建

4、环境搭建：基于Docker搭建Flink开发环境，配置Kafka消息队列，安装Python SDK

好，咱们正式开始动手了。

这一章，说白了就是搭积木。把 Flink、Kafka、Python SDK 这三块核心积木拼到一起，让它们能互相通信。我见过太多同学在环境上卡一整天，最后发现是某个端口没开，或者版本不匹配。所以咱们一步步来，稳一点。

4.1 为什么选择 Docker？

你可能会问：为啥非要用 Docker？直接在本地装 Flink 和 Kafka 不行吗？

行，当然行。但我个人习惯用 Docker，原因有三：

隔离性：不会把你本机的 Java、Python 环境搞得一团糟。我之前有一次装 Flink，结果把系统自带的 Python 版本给覆盖了，修了半天。
一致性：你在 Mac 上搭的环境，拿到 Linux 服务器上也能跑。团队协作时，大家拉同一个镜像，省去「我这儿能跑啊」的扯皮。
快速重置：玩坏了？删掉容器，重新 docker-compose up，一分钟恢复如初。

小提示：如果你对 Docker 还不太熟，建议先花 10 分钟看看 docker-compose 的基本语法。咱们后面会用到的。

4.2 整体架构图

先上一张图，让你对整体结构有个直观印象。这张图我手绘的 SVG，虽然丑了点，但逻辑清楚。

嗯，这张图的核心逻辑就是：Python 模拟数据 -> Kafka 中转 -> Flink 实时计算 -> 结果输出。咱们后面所有的实战代码，都跑在这个架构上。

4.3 第一步：编写 docker-compose.yml

这是整个环境的「总开关」。我习惯把所有服务写在一个文件里，方便管理。

version: '3.8'

services:
  # ---------- Zookeeper（Kafka 依赖它） ----------
  zookeeper:
    image: confluentinc/cp-zookeeper:7.4.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - "2181:2181"

  # ---------- Kafka 消息队列 ----------
  kafka:
    image: confluentinc/cp-kafka:7.4.0
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  # ---------- Flink JobManager ----------
  flink-jobmanager:
    image: flink:1.17.1-scala_2.12
    ports:
      - "8081:8081"
    command: jobmanager
    environment:
      - JOB_MANAGER_RPC_ADDRESS=flink-jobmanager

  # ---------- Flink TaskManager ----------
  flink-taskmanager:
    image: flink:1.17.1-scala_2.12
    depends_on:
      - flink-jobmanager
    command: taskmanager
    environment:
      - JOB_MANAGER_RPC_ADDRESS=flink-jobmanager
      - TASK_MANAGER_NUMBER_OF_TASK_SLOTS=2

注意：Kafka 的 ADVERTISED_LISTENERS 一定要配成 localhost:9092。我曾经在这里踩过坑——配成容器内部地址，结果 Python 客户端连不上，排查了半小时才发现是地址问题。

4.4 第二步：启动环境

文件写好了，启动就一行命令：

docker-compose up -d

等个一两分钟，看看所有容器是不是都起来了：

docker-compose ps

你应该能看到类似这样的输出：

服务名	状态	端口
zookeeper	Up	2181
kafka	Up	9092
flink-jobmanager	Up	8081
flink-taskmanager	Up	-

打开浏览器，访问 http://localhost:8081，你应该能看到 Flink 的 Web UI。看到那个界面，心里就踏实了。

4.5 第三步：在 Kafka 中创建 Topic

Kafka 启动后，咱们得先建个 Topic，不然数据往哪儿发？

docker exec -it <kafka容器名> kafka-topics --create \
  --topic risk_events \
  --bootstrap-server localhost:9092 \
  --partitions 3 \
  --replication-factor 1

验证一下 Topic 是否创建成功：

docker exec -it <kafka容器名> kafka-topics --list --bootstrap-server localhost:9092

如果看到 risk_events 出现在列表里，说明成了。

小技巧：分区数设成 3，是为了后面演示 Flink 的并行消费。你想想看，如果只有一个分区，那并行度再高也白搭。

4.6 第四步：安装 Python SDK

Python 这边，我们需要两个库：

kafka-python：用来往 Kafka 发消息
flink-py（或者用 PyFlink）：用来写 Flink 作业

安装命令：

pip install kafka-python apache-flink==1.17.1

注意版本号。Flink 1.17 对应 PyFlink 1.17，别搞混了。我之前有个同事装了 1.14 的 PyFlink，结果连 Flink 1.17 的集群，各种报错。

验证安装：

python -c "from kafka import KafkaProducer; print('Kafka SDK OK')"
python -c "from pyflink.datastream import StreamExecutionEnvironment; print('PyFlink OK')"

如果没报错，恭喜你，环境搭好了。

4.7 快速测试：发一条消息试试

写个简单的 Python 脚本，往 Kafka 里发一条测试消息：

from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

test_event = {
    "user_id": "u001",
    "action": "login",
    "ip": "192.168.1.1",
    "timestamp": 1700000000
}

producer.send('risk_events', test_event)
producer.flush()
print("消息发送成功")

跑一下这个脚本，如果没报错，说明 Python 到 Kafka 的链路通了。

  核心要点：环境搭建的本质，就是让「数据源 -> Kafka -> Flink」这条链路跑通。后面的所有风控逻辑，都建立在这个基础上。

好，环境这块就到这儿。咱们把 Flink、Kafka、Python SDK 都装好了，也验证了通信。下一章，我会带你写第一个真正的风控规则——用 Flink 实时检测「同一 IP 短时间内多次登录」的异常行为。

公众号:蓝海资料掘金营，微信deep3321

📚 课程目录

4、环境搭建：基于Docker搭建Flink开发环境，配置Kafka消息队列，安装Python SDK

4.1 为什么选择 Docker？

4.2 整体架构图

4.3 第一步：编写 docker-compose.yml

4.4 第二步：启动环境

4.5 第三步：在 Kafka 中创建 Topic

4.6 第四步：安装 Python SDK

4.7 快速测试：发一条消息试试