流式数据处理入门：Kafka 与实时计算——AI 数据管道的动脉

如果业务要求事件发生后尽快更新结果，每天一次的批处理就可能来不及。Kafka 可以保存和传递事件，计算仍需要消费者或流处理引擎完成。下面先讲生产、分区和消费，再讨论重复消息、乱序和重放时怎样保持结果一致。

这篇从 AI 工程师视角讲 Kafka：核心概念只讲和实战相关的，Python 生产者消费者代码直接可用，语义保障（at-least-once / exactly-once）讲清楚取舍，最后落到「实时特征管道」这个 AI 场景。

前置阅读：建议先读 SQL 与数据库实战、Linux + Python 环境基础。了解 大数据管理里的批处理后再看流式，对比更鲜明。

为什么 AI 项目需要流式

三个真实场景，我都接过：

1. 实时特征：推荐模型要用「用户最近 10 分钟点击了 3 次数码类商品」这种特征，等凌晨批处理早凉了。
2. 模型监控：线上预测结果实时落流，计算延迟/分布漂移指标，异常立刻告警，而不是等 T+1 报表才发现模型挂了。
3. 数据同步：业务库的变更（CDC）实时流到数据湖和搜索引擎，保证 RAG 知识库分钟级更新。

批处理是「仓库盘点」，流处理是「收银台流水」。两者互补，谁也别想取代谁。

Kafka 核心概念：只讲用得上的

Kafka 本质是一个分布式、持久化的发布订阅日志。五个概念构成全部心智模型：

Producer ──→ Topic（主题，逻辑分类）
                └── Partition（分区，并行单位，物理存储）
                      └── Offset（每条消息在分区里的递增编号）
Consumer ──→ Consumer Group（消费组，同组内分区互斥分配）

关键理解点：

• 分区是并行度的天花板：一个 topic 8 个分区，同一个消费组里最多 8 个消费者同时干活，第 9 个干瞪眼。所以分区数要按吞吐规划，多了不能随便减（只能增）。
• offset 由消费者自己保管（提交到 Kafka 内部 topic），消费到哪了是消费者的责任——这决定了「重复消费」和「丢消息」是设计选项，不是 bug。
• 消息保留期与消费无关：消息存 7 天（默认），消不消费它都在。和「读完即删」的传统 MQ（RabbitMQ 语义）完全不同——Kafka 更像一个可被反复读的日志存储。

本地起一个单节点 Kafka（KRaft 模式，不依赖 Zookeeper）：

docker run -d --name kafka -p 9092:9092 \
  -e KAFKA_NODE_ID=1 \
  -e KAFKA_PROCESS_ROLES=broker,controller \
  -e KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \
  -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
  -e KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:9093 \
  apache/kafka:3.8.0

生产者/消费者实战（confluent-kafka）

from confluent_kafka import Producer, Consumer, KafkaError
import json

# ---------- 生产者 ----------
producer = Producer({"bootstrap.servers": "localhost:9092"})

def delivery_report(err, msg):
    if err:
        print(f"投递失败: {err}")

event = {"user_id": 10086, "item_id": "SKU_9527",
         "action": "click", "ts": 1756444800}
producer.produce(
    "user_events",
    key=str(event["user_id"]).encode(),   # key 决定分区：同 key 进同分区，保序！
    value=json.dumps(event).encode(),
    callback=delivery_report,
)
producer.flush()   # 阻塞到全部投递完成

key 的选择是设计决策：同一个 user_id 的事件进同一个分区，就保证了「同一用户的行为有序」。这是用户级会话特征能算对的前提。不设 key 则轮询分发，吞吐高但全局无序。

# ---------- 消费者 ----------
consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "feature_computing",
    "auto.offset.reset": "earliest",     # 新组从头读
    "enable.auto.commit": False,          # 手动提交，自己掌控语义
})
consumer.subscribe(["user_events"])

while True:
    msg = consumer.poll(1.0)
    if msg is None:
        continue
    if msg.error():
        continue
    event = json.loads(msg.value())
    process(event)                        # 业务处理
    consumer.commit(msg)                  # 处理成功后再提交 offset

三种语义：重复、丢失、精确，选一个

消息系统的经典三难，由「offset 提交时机」和「生产端确认」共同决定：

语义	配置	结果
At-most-once（可能丢）	先提交 offset 再处理	处理崩了消息就丢了
At-least-once（可能重复）	先处理再提交 offset	提交前崩溃 → 重投 → 重复消费
Exactly-once	事务 + 幂等生产者	精确一次，代价是吞吐降 20~30%

工程界的共识答案：at-least-once + 消费端幂等。让你的 process(event) 幂等（同一事件处理多次结果相同）比上事务便宜得多。幂等的标准做法：事件带唯一 event_id，处理前先查「这个 id 处理过吗」（Redis SETNX 或库表唯一约束），处理过直接跳过。

我犯过的真实错误：消费逻辑里「处理 → 更新数据库 → 提交 offset」三步没有原子性，处理完数据库成功、提交前进程被 kill，重启后重放导致数据库里计数翻倍。后来加 event_id 去重表解决。任何不做幂等的消费者都是定时炸弹。

消费者组：扩缩容与再平衡

同一个 group 里加消费者实例，Kafka 自动重新分配分区（rebalance）。这是水平扩容的钥匙：流量涨了，多开几个消费进程就行。

但 rebalance 有暗坑：再平衡期间整个组停止消费（stop-the-world），且会重复消费已拉取未提交的消息。频繁 rebalance（消费者因 GC 或心跳超时进进出出）是生产事故高发区。缓解：增大 session.timeout.ms、处理逻辑别在 poll 循环里干重活（超过 max.poll.interval.ms 默认 5 分钟会被踢出组）。

流式计算：Kafka 之上的一层

只消费原始事件做简单加工，Consumer API 够了。但「最近 10 分钟窗口内的点击次数」「双流 JOIN」这类有状态计算，手写状态管理会疯——这时候上流计算框架：

• Flink：流处理的事实标准，事件时间语义、水位线、精确一次状态、SQL 支持。公司有平台团队就用它。
• Kafka Streams：Kafka 官方轻量库，嵌入 Java 应用，无需独立集群。中小规模首选。
• Faust（Python）：Kafka Streams 的 Python 影子，社区活跃度下降，慎用于新项目。

用 Python 手写一个「滑动窗口计数」也能理解原理（Faust 风格）：

from collections import defaultdict, deque
import time

class SlidingWindowCounter:
    """每个 key 维护最近 window_sec 秒的事件计数"""
    def __init__(self, window_sec=600):
        self.window = window_sec
        self.events = defaultdict(deque)   # key -> deque[timestamp]

    def add(self, key, ts=None):
        ts = ts or time.time()
        q = self.events[key]
        q.append(ts)
        while q and ts - q[0] > self.window:   # 弹出过期
            q.popleft()
        return len(q)

counter = SlidingWindowCounter(window_sec=600)
# 消费循环里：
#   cnt = counter.add(f"{user_id}:{category}")
#   feature_store.write(user_id, f"click_10min_{category}", cnt)

这个几十行的实现就是「实时特征」的最小内核。生产环境里它换成 Flink 的 TumblingEventTimeWindow + RocksDB 状态后端，思想完全一样：事件驱动、维护状态、输出特征。

搭建一条最小实时管道

把整篇串起来的完整链路（推荐系统实时特征场景）：

App 埋点 → Kafka topic: user_events（8 分区）
             ├── 消费者组 A：实时特征计算 → Redis（特征在线读取）
             ├── 消费者组 B：落数据湖（Parquet）→ 次日训练用
             └── 消费者组 C：异常检测 → 告警

注意架构精髓：同一条流被多个消费组各自独立消费，互不干扰。新需求来了（比如加个审计落库），新起个消费组订阅即可，不用动任何现有系统。这是 Kafka 发布订阅模式最大的架构红利——解耦。

另一个经验：事件 schema 要有 event_id、event_time、schema_version 三个字段。没有 event_time 用处理时间做窗口，数据迟到就全乱；没有 schema_version，字段演进时新老消费者互相踩。

踩坑排查

现象	原因	解法
消费者收不到消息也不报错	group.id 被别人复用，offset 已在末尾	换新 group 名测试；确认 auto.offset.reset
重复消费爆发	处理慢超 max.poll.interval 被踢组 → rebalance	处理逻辑轻量化，或增大间隔 + 减小批量
分区消费不均（有的撑死有的饿死）	key 倾斜（大 V 用户事件占一个分区）	key 加随机后缀打散，业务层再聚合
生产者发送慢	逐条同步发送	批量 linger.ms=10 + batch.size 调大，异步回调
消息顺序错乱	没设 key 或重试导致乱序	同 key 保序；max.in.flight=1 保严格序（牺牲吞吐）
磁盘暴涨	保留期 retention.ms 过长 + 流量大	按容量规划保留期，或开 compact（只留每 key 最新值）

练习

1. 本地起 Kafka，写生产者每秒生成 10 条模拟用户行为，消费者统计「每秒各品类点击量」打印出来。
2. 给消费者加幂等：用 Redis 记录已处理的 event_id，模拟进程重启（kill 后重跑）验证无重复计数。
3. 实现文中的 SlidingWindowCounter，用模拟数据验证：暂停生产 10 分钟后恢复，窗口计数正确回落。
4. 设计题：给 RAG 项目加「知识库实时更新」——文档入库事件流经 Kafka，消费者自动触发切分、embedding、写向量库。画出组件图。

面试常问

Q：Kafka 为什么高吞吐？ 四个设计：① 顺序写磁盘（append-only log，磁盘顺序写堪比内存随机写）；② Page Cache 读写基本不经过 JVM 堆；③ 零拷贝（sendfile，数据不经过用户态直接网卡）；④ 批量 + 压缩攒批传输。单机百万级 TPS 的来源。

Q：Kafka 和 RabbitMQ 怎么选？ RabbitMQ 是传统消息队列：灵活路由（exchange）、消息确认到条、读完即删，适合「任务分发、RPC 解耦」。Kafka 是流平台：日志语义、海量堆积、可重放、多订阅方，适合「数据管道、事件溯源、流计算」。要重放历史数据选 Kafka，要复杂路由选 RabbitMQ。

Q：分区数怎么定？ 由目标吞吐和单分区能力反推：单分区消费大约 10MB/s 量级，要 100MB/s 就 ≥10 分区；同时考虑消费者并行度和未来扩容（分区只增不减，一次给足富余）。不是越多越好——分区多了元数据和文件句柄开销上升，上万分区 controller 压力大。

Q：exactly-once 怎么实现，为什么大家少用？ Kafka 的 EOS = 幂等生产者（sequence number 去重）+ 事务（跨分区原子写 + offset 与业务结果同一事务提交）。代价：吞吐降、延迟升、运维复杂。而「at-least-once + 消费端幂等」用很小的业务成本达到等效效果，所以工程界压倒性选后者。EOS 主要在流计算框架内部状态一致性上用（Flink 两阶段提交对接 Kafka）。

Q：数据倾斜怎么处理？ 识别：监控各分区 lag 和消息量差异。处理：① key 设计打散（user_id + 随机后缀）；② 两级聚合（先局部聚合再全局合并）；③ 热点 key 单独处理（大 V 单独逻辑）。源头思维：倾斜是业务本质（二八分布），工程上只能稀释不能消灭。

相关阅读

• 大数据管理：Hadoop、Spark 与数据仓库——批处理一侧的对照
• NoSQL 选型实战——实时特征的归宿 Redis 在这里
• SQL 与数据库实战——CDC 的上游
• 时间序列分析实战——流式监控指标的下游分析
• 推荐系统入门——实时特征最终服务的模型

流式处理的心智转变就一句话：从「数据在那里，我去查」变成「数据在流动，我在河边接住它」。想通这个，Kafka 的所有设计都顺理成章。