📨 05 · Kafka 与数据管道：分区 / 偏移量 / 消费者组

%matplotlib inline
import matplotlib
import matplotlib.pyplot as plt
matplotlib.rcParams['font.sans-serif'] = ['Microsoft YaHei', 'SimHei', 'PingFang SC', 'Noto Sans CJK SC', 'Arial Unicode MS']
matplotlib.rcParams['axes.unicode_minus'] = False
import numpy as np
import pandas as pd
import sqlite3
import os
rng = np.random.default_rng(42)

print('''Kafka 组件：
  Topic  ：逻辑消息类别（如 user_click）；
  Partition：Topic 的物理分片（并行单位，一个分区内有序）；
  Offset ：分区内消息的序号（从 0 开始）；
  Broker ：服务器节点（存分区副本）；
  Producer / Consumer：生产者 / 消费者；
  Consumer Group：一组消费者协作消费一个 Topic（组内互斥、组间广播）。

为什么用 Kafka：解耦（上游不用管下游）、削峰（缓冲）、
              顺序性（分区内）、持久化（落盘可回放）。
''')

Kafka 组件：
  Topic  ：逻辑消息类别（如 user_click）；
  Partition：Topic 的物理分片（并行单位，一个分区内有序）；
  Offset ：分区内消息的序号（从 0 开始）；
  Broker ：服务器节点（存分区副本）；
  Producer / Consumer：生产者 / 消费者；
  Consumer Group：一组消费者协作消费一个 Topic（组内互斥、组间广播）。

为什么用 Kafka：解耦（上游不用管下游）、削峰（缓冲）、
              顺序性（分区内）、持久化（落盘可回放）。

import zlib
# 注：Python 内置 hash() 对字符串带随机盐，每次运行结果不同；
# 教学演示改用确定性哈希 zlib.crc32，保证结果可复现（Kafka 默认用 murmur2，思路相同：确定性哈希取模）
n_part = 3
msgs = [('u1', 'click'), ('u2', 'view'), ('u1', 'buy'), ('u3', 'click'), ('u2', 'search'), ('u1', 'view')]
partitions = [[] for _ in range(n_part)]
for key, val in msgs:
    p = zlib.crc32(key.encode()) % n_part
    partitions[p].append((key, val, len(partitions[p])))   # (key, val, offset)
for i, p in enumerate(partitions):
    print(f'分区{i}: {p}')
print('\n➡️ 相同 key（同用户）会被路由到同一分区 -> 该用户消息有序');
assert all(len(p) >= 0 for p in partitions)
print('✅ 分区路由演示完成')

分区0: [('u2', 'view', 0), ('u2', 'search', 1)]
分区1: []
分区2: [('u1', 'click', 0), ('u1', 'buy', 1), ('u3', 'click', 2), ('u1', 'view', 3)]

➡️ 相同 key（同用户）会被路由到同一分区 -> 该用户消息有序
✅ 分区路由演示完成

# 消费者组：3 个消费者平分 3 个分区（每个消费者一个分区）
consumers = {f'consumer-{i}': [i] for i in range(n_part)}
# 消费进度：记录每个分区已消费的 offset
offsets = {i: 0 for i in range(n_part)}
consumed = {f'consumer-{i}': [] for i in range(n_part)}

# 模拟消费：各消费者从自己分区拉取
while True:
    progressed = False
    for i, p in enumerate(partitions):
        cname = f'consumer-{i}'
        if offsets[i] < len(p):
            key, val, ofs = p[offsets[i]]
            consumed[cname].append(val)
            offsets[i] += 1
            progressed = True
    if not progressed:
        break

for c, v in consumed.items():
    print(f'{c}: 消费 {v} （该分区已消费 offset {offsets[int(c.split("-")[1])]}）')
print('\n➡️ 同一用户消息保持顺序；组内消费者各管各的分区，进度记录在 offset')

consumer-0: 消费 ['view', 'search'] （该分区已消费 offset 2）
consumer-1: 消费 [] （该分区已消费 offset 0）
consumer-2: 消费 ['click', 'buy', 'click', 'view'] （该分区已消费 offset 4）

➡️ 同一用户消息保持顺序；组内消费者各管各的分区，进度记录在 offset

# 可视化：每个分区的消费进度（已提交 offset vs 最新 offset）
fig, ax = plt.subplots(figsize=(9.5, 3.6))
for i, p in enumerate(partitions):
    newest = len(p)
    ax.barh(i, newest + 1, color='#F2F2F2', ec='#7F7F7F', height=0.5)
    ax.barh(i, max(offsets[i], 0.001), color='#548235', height=0.5)
    ax.text(newest / 2, i, f'分区{i}: 最新offset={newest}, 已提交={offsets[i]}, 积压={newest - offsets[i]}',
            ha='center', va='center', fontsize=9.5, color='black')
ax.set_yticks(range(n_part)); ax.set_yticklabels([f'分区{i}' for i in range(n_part)])
ax.set_xlabel('消息位置（offset 从 0 递增）')
ax.set_title('消费进度：绿色=已提交位置；灰色剩余=未消费积压', fontsize=12)
plt.show()

# 5 个分区、3 个消费者时的分配（简化 round-robin）
n_part5, n_cons3 = 5, 3
assign = {f'c{i}': [] for i in range(n_cons3)}
for p in range(n_part5):
    assign[f'c{p % n_cons3}'].append(p)
print('5 分区 / 3 消费者 分配:')
for c, ps in assign.items():
    print(f'  {c}: 分区 {ps}')
print('''
  消费者加入/退出 -> 触发 Rebalance（分区重新分配）；
  代价：rebalance 期间消费暂停、可能重复消费（需幂等/去重）。
''')

5 分区 / 3 消费者 分配:
  c0: 分区 [0, 3]
  c1: 分区 [1, 4]
  c2: 分区 [2]

  消费者加入/退出 -> 触发 Rebalance（分区重新分配）；
  代价：rebalance 期间消费暂停、可能重复消费（需幂等/去重）。

print('''生产者 acks 三档：
  acks=0：不等确认（最快，可能丢）；
  acks=1：leader 落盘即确认（默认，丢 leader 时可能丢）；
  acks=all：全部 ISR 副本确认（最稳，慢）。

消费者语义：
  At-most-once ：可能丢（offset 先提交）；
  At-least-once：可能重（offset 后提交 + 重试）；
  Exactly-once ：幂等写入 + 事务（Kafka 支持）。

可靠性组合拳：3 副本 + acks=all + 消费端幂等去重。
''')

生产者 acks 三档：
  acks=0：不等确认（最快，可能丢）；
  acks=1：leader 落盘即确认（默认，丢 leader 时可能丢）；
  acks=all：全部 ISR 副本确认（最稳，慢）。

消费者语义：
  At-most-once ：可能丢（offset 先提交）；
  At-least-once：可能重（offset 后提交 + 重试）；
  Exactly-once ：幂等写入 + 事务（Kafka 支持）。

可靠性组合拳：3 副本 + acks=all + 消费端幂等去重。

print('''1) 分区顺序 vs 全局顺序？ -> 只保证分区内有序，要全局有序只能 1 分区（牺牲并行）；
2) 消息积压怎么办？       -> 扩分区+扩消费者、批量消费、临时降级；
3) 消费速度跟不上？       -> 增加消费者（≤分区数）、优化反序列化、批量；
4) 数据重复消费？         -> 幂等（唯一键去重）或事务；
5) Kafka vs RocketMQ？    -> 吞吐/顺序/事务侧重不同；
6) 为什么高吞吐？         -> 顺序写磁盘、零拷贝、批量/压缩。
''')

1) 分区顺序 vs 全局顺序？ -> 只保证分区内有序，要全局有序只能 1 分区（牺牲并行）；
2) 消息积压怎么办？       -> 扩分区+扩消费者、批量消费、临时降级；
3) 消费速度跟不上？       -> 增加消费者（≤分区数）、优化反序列化、批量；
4) 数据重复消费？         -> 幂等（唯一键去重）或事务；
5) Kafka vs RocketMQ？    -> 吞吐/顺序/事务侧重不同；
6) 为什么高吞吐？         -> 顺序写磁盘、零拷贝、批量/压缩。