数据模拟层开发：让数据先跑起来

好，到了第四章，我们终于要开始写代码了。

前面三章我们把架构、技术选型、数据模型都聊完了。但有个问题——平台搭好了，数据从哪来？你总不能用生产环境的数据做测试吧？我记得我第一次搭实时平台时，直接连了线上Kafka，结果把业务topic搞乱了，被运维追着骂了三天。

所以，我们需要一个数据模拟层。说白了，就是自己造数据。

4.1 用户行为日志的JSON格式定义

先定协议。我们模拟的是用户行为日志，每条日志就是一个JSON对象。我个人习惯把字段分成三类：基础标识、行为信息、上下文环境。

你可能会问：为什么不用Avro或Protobuf？嗯，在模拟层用JSON有两个好处：一是肉眼可读，调试方便；二是后面接入Flink时，JSON解析的灵活性更高。等上了生产，再考虑序列化压缩的事。

下面是我常用的日志生成函数：

import json
import random
import time
from datetime import datetime

def generate_user_event():
    event = {
        "user_id": f"u_{random.randint(10000, 99999)}",
        "event_type": random.choice(["page_view", "click", "purchase", "add_cart"]),
        "page_id": random.choice(["p_home", "p_detail", "p_cart", "p_order"]),
        "duration_ms": random.randint(500, 15000),
        "timestamp": int(time.time() * 1000),
        "ip": f"192.168.{random.randint(0,255)}.{random.randint(1,254)}",
        "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
    }
    return json.dumps(event)

4.2 Kafka Producer：把数据发出去

数据造好了，得发到Kafka里去。Python连Kafka，我推荐用confluent_kafka库。为什么不用kafka-python？因为confluent_kafka底层是C写的，性能好得多。我在一次压测中，同样的机器配置，confluent_kafka的吞吐量高出3倍。

先装依赖：

pip install confluent_kafka

然后写Producer：

from confluent_kafka import Producer
import json

conf = {
    'bootstrap.servers': 'localhost:9092',
    'acks': '1',
    'retries': 3,
    'batch.size': 16384,
    'linger.ms': 10
}

producer = Producer(conf)

def delivery_report(err, msg):
    if err is not None:
        print(f'发送失败: {err}')
    else:
        print(f'发送成功: {msg.topic()} [{msg.partition()}]')

def send_event(event_json):
    producer.produce(
        topic='user_behavior',
        value=event_json.encode('utf-8'),
        callback=delivery_report
    )
    producer.poll(0)

4.3 数据速率控制：别把Kafka打爆

模拟数据最怕什么？怕你一口气发太多，把Kafka搞挂了。我见过有人写了个死循环，每秒发几万条，直接把测试集群的磁盘写满了。

所以，我们要做速率控制。常用的方法有两种：

• 固定速率：每秒发N条，用time.sleep控制间隔
• 动态速率：模拟真实流量，比如白天快、晚上慢

先看固定速率的实现：

import time

def run_fixed_rate(rate_per_second, duration_seconds):
    """
    rate_per_second: 每秒发送条数
    duration_seconds: 持续秒数
    """
    interval = 1.0 / rate_per_second
    start = time.time()
    count = 0
    
    while time.time() - start < duration_seconds:
        event = generate_user_event()
        send_event(event)
        count += 1
        time.sleep(interval)
    
    print(f"共发送 {count} 条数据")

动态速率稍微复杂一点。我习惯用正弦波模拟日周期：

import math

def run_dynamic_rate(base_rate=100, peak_rate=500, cycle_minutes=60):
    """
    base_rate: 最低速率
    peak_rate: 最高速率
    cycle_minutes: 一个完整周期(分钟)
    """
    start = time.time()
    
    while True:
        elapsed = (time.time() - start) / 60.0  # 转为分钟
        # 用sin函数模拟流量波动
        rate = base_rate + (peak_rate - base_rate) * (math.sin(elapsed * 2 * math.pi / cycle_minutes) + 1) / 2
        
        interval = 1.0 / rate
        event = generate_user_event()
        send_event(event)
        time.sleep(interval)

4.4 完整的数据模拟流程

把上面几块拼起来，就是完整的数据模拟层。我画了张图，帮你理清关系：

流程其实很简单：生成JSON → 控制速率 → 发送到Kafka。但每个环节都有坑，我帮你列一下：

• JSON生成：字段类型要对，尤其是时间戳和数值型，别发字符串
• 速率控制：别用死循环，加个退出条件（比如按Ctrl+C优雅退出）
• Kafka发送：一定要处理回调，否则发没发出去你都不知道

4.5 避坑指南：我踩过的三个坑

最后，分享几个真实教训：

1. 消息体太大：有一次我把整个页面HTML塞进了日志，单条消息超过1MB，Kafka直接报错。后来限制了单条消息不超过100KB。
2. 分区不均匀：如果不指定key，Kafka默认轮询分区。但如果你用user_id做key，热点用户会导致数据倾斜。模拟时建议不加key，让分区均匀分布。
3. 忘记flush：程序退出前一定要调producer.flush()，否则缓冲区里的消息会丢失。我吃过这个亏，跑了半小时的数据，一关程序全没了。