第12课：Kafka入门

📡 Kafka入门

Apache Kafka是分布式流处理平台，以高吞吐、低延迟、持久化著称，是现代数据管道的事实标准。

🐍 Python实战：核心实战

# Kafka模拟引擎
import hashlib, time, random, json
from datetime import datetime
from collections import defaultdict

class Partition:
    def __init__(self, id): self.id = id; self.messages = []; self.offset = 0
    def append(self, msg): self.messages.append({'offset':self.offset,'message':msg,'timestamp':datetime.now().isoformat()}); self.offset += 1
    def read(self, start_offset=0):
        return [m for m in self.messages if m['offset'] >= start_offset]

class KafkaTopic:
    def __init__(self, name, num_partitions=3):
        self.name = name; self.partitions = [Partition(i) for i in range(num_partitions)]
    def produce(self, key, value):
        pid = int(hashlib.md5(str(key).encode()).hexdigest(),16) % len(self.partitions)
        self.partitions[pid].append({'key':key,'value':value})
        return pid
    def consume(self, group_id, partitions=None):
        results = []
        for p in (self.partitions if partitions is None else [self.partitions[i] for i in partitions]):
            results.extend(p.read())
        return results
    def stats(self):
        return {'partitions':len(self.partitions),'messages':sum(len(p.messages) for p in self.partitions),
                'distribution':[len(p.messages) for p in self.partitions]}

class KafkaProducer:
    def __init__(self): self.sent = 0
    def send(self, topic, key, value): pid = topic.produce(key, value); self.sent += 1; return pid

class KafkaConsumer:
    def __init__(self, group_id): self.group_id = group_id; self.offsets = defaultdict(int)
    def poll(self, topic):
        results = topic.consume(self.group_id)
        new = [m for m in results if m['offset'] >= self.offsets[topic.name]]
        if new: self.offsets[topic.name] = max(m['offset'] for m in new) + 1
        return new

# 测试
topic = KafkaTopic("orders", 3)
producer = KafkaProducer()
for i in range(1000):
    producer.send(topic, f"ORD_{i:05d}", {'amount':round(random.uniform(10,5000),2),'region':random.choice(['华东','华南','华北'])})

print(f"Kafka模拟: {topic.stats()}")
print(f"生产者: {producer.sent}条")

consumer1 = KafkaConsumer("group-1")
consumer2 = KafkaConsumer("group-2")
msgs1 = consumer1.poll(topic)
msgs2 = consumer2.poll(topic)
print(f"消费者group-1: {len(msgs1)}条")
print(f"消费者group-2: {len(msgs2)}条(独立消费)")

print("\n分区分布:")
for i, p in enumerate(topic.partitions):
    print(f"  分区{i}: {len(p.messages)}条")

print("✅ 验证通过 - Kafka模拟运行正常")

📖 原理深入：流处理核心理论

🏆 最佳实践：流处理设计检查清单

🔬 深度案例：生产环境实战经验

📚 延伸阅读与参考

🤔 常见问题与解答

🔑 关键技术选型决策

数据量级？
├── <1GB/天 → SQLite/PostgreSQL + Python脚本
├── 1-100GB/天 → 数据仓库(Snowflake/BQ) + dbt
├── 100GB-1TB/天 → Spark + 数据湖
└── >1TB/天 → Spark/Flink + 湖仓一体

延迟要求？
├── T+1 → 批处理(Spark) + Airflow
├── 小时级 → 微批(Spark Streaming) + Airflow
├── 分钟级 → 流处理(Flink) + Kafka
└── 秒级 → 纯流(Flink) + Kafka + 缓存

数据类型？
├── 纯结构化 → 数据仓库
├── 结构化+半结构化 → 湖仓一体
└── 全类型 → 数据湖 + 元数据管理