📚 课程目录第3章数据模拟与采集层

数据模拟与采集层：从零搭建数据高速公路

数据是实时数仓的血液。没有数据，再牛的架构也是空中楼阁。这一章，我们就来解决「数据从哪来」的问题。

很多同学觉得数据模拟很简单，不就是造点假数据嘛。但我在几个大厂项目里踩过的坑告诉我——数据模拟的质量，直接决定了后续开发调试的效率。你想想看，如果模拟数据跟真实业务对不上，你辛辛苦苦搭好的实时链路，一上线就崩，那得多尴尬。

数据埋点设计：一切从业务出发

埋点，说白了就是告诉系统「用户在干什么」。我见过太多团队一上来就堆埋点，结果数据量爆炸，真正有用的指标反而拿不到。

我个人习惯，先画一张用户行为流转图。比如一个电商App，用户从打开App到下单，会经历哪些步骤？

核心埋点事件清单（电商场景）

启动事件：app_launch — 记录启动时间、渠道来源
浏览事件：page_view — 页面ID、停留时长、商品ID（如果有）
搜索事件：search — 关键词、搜索结果数
加购事件：add_to_cart — 商品ID、数量、价格
下单事件：order_create — 订单ID、商品列表、总金额
支付事件：payment — 支付方式、支付状态

每个事件都包含公共属性和业务属性。公共属性比如设备ID、用户ID、时间戳、网络环境；业务属性则跟具体事件相关。

我的经验：埋点字段不要贪多。我曾经在一个项目里，一个事件塞了50多个字段，结果一半都是空的。后来我们定了个规矩——每个事件不超过20个字段，够用就行。

模拟数据生成器开发：Java vs Python

数据模拟器，我两种语言都写过。这里说说我的感受。

对比维度	Java	Python
开发速度	中等	快（库多，代码少）
性能	高（适合高吞吐）	一般（GIL限制）
与Kafka集成	原生支持（Kafka Client）	需要额外库（kafka-python）
适用场景	生产级压测、高并发模拟	快速原型、小规模测试

我个人建议：课程项目用Python快速出原型，生产环境用Java做压测。下面我给出两个版本的代码片段，你感受一下差异。

Python版本（快速上手）

import json
import time
import random
from datetime import datetime
from kafka import KafkaProducer

# 模拟用户行为
def generate_event():
    event_types = ['app_launch', 'page_view', 'search', 'add_to_cart', 'order_create', 'payment']
    event = {
        "event_id": int(time.time() * 1000),
        "event_type": random.choice(event_types),
        "user_id": f"user_{random.randint(1, 10000)}",
        "device_id": f"device_{random.randint(1, 5000)}",
        "timestamp": datetime.now().isoformat(),
        "properties": {}
    }
    # 根据事件类型填充业务属性
    if event['event_type'] == 'page_view':
        event['properties']['page_id'] = f"page_{random.randint(1, 50)}"
        event['properties']['stay_seconds'] = random.randint(1, 300)
    elif event['event_type'] == 'order_create':
        event['properties']['order_id'] = f"order_{int(time.time())}"
        event['properties']['amount'] = round(random.uniform(10, 5000), 2)
    return event

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

while True:
    event = generate_event()
    producer.send('ods_events_raw', value=event)
    time.sleep(random.uniform(0.1, 1))  # 模拟随机间隔

Java版本（生产级）

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import com.alibaba.fastjson.JSONObject;
import java.util.Properties;
import java.util.Random;

public class DataSimulator {
    private static final String[] EVENT_TYPES = {"app_launch", "page_view", "search", "add_to_cart", "order_create", "payment"};
    private static final Random RANDOM = new Random();

    public static JSONObject generateEvent() {
        JSONObject event = new JSONObject();
        event.put("event_id", System.currentTimeMillis() + RANDOM.nextInt(1000));
        event.put("event_type", EVENT_TYPES[RANDOM.nextInt(EVENT_TYPES.length)]);
        event.put("user_id", "user_" + RANDOM.nextInt(10000));
        event.put("device_id", "device_" + RANDOM.nextInt(5000));
        event.put("timestamp", System.currentTimeMillis());

        JSONObject props = new JSONObject();
        String eventType = event.getString("event_type");
        if ("page_view".equals(eventType)) {
            props.put("page_id", "page_" + RANDOM.nextInt(50));
            props.put("stay_seconds", RANDOM.nextInt(300) + 1);
        } else if ("order_create".equals(eventType)) {
            props.put("order_id", "order_" + System.currentTimeMillis());
            props.put("amount", Math.round(RANDOM.nextDouble() * 5000 * 100.0) / 100.0);
        }
        event.put("properties", props);
        return event;
    }

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        while (true) {
            JSONObject event = generateEvent();
            ProducerRecord<String, String> record = new ProducerRecord<>("ods_events_raw", event.toJSONString());
            producer.send(record);
            try {
                Thread.sleep(RANDOM.nextInt(1000) + 100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

避坑指南：我曾经在生产环境用Python模拟器做压测，结果发现单机只能跑到每秒500条，CPU就满了。后来换成Java版本，同样的机器跑到每秒5000条毫无压力。所以——压测场景一定要用Java。

Kafka Topic设计：命名即规范

Topic设计看似简单，但命名不规范，后期运维会非常痛苦。我见过一个项目，Topic名字叫「test」「data」「log」，三个月后没人知道哪个Topic对应哪个业务。

我推荐一套命名规范：

层级_数据域_业务过程_分区策略
例如：ods_events_raw_12p
     dwd_events_clean_6p
     ads_order_daily_3p

层级前缀	含义	分区数建议	副本数
ods_	原始数据层	12	2
dwd_	明细数据层	6	2
dws_	汇总数据层	3	2
ads_	应用数据层	3	2

为什么分区数这么设计？嗯，这里有个小技巧——分区数最好跟消费者并行度匹配。比如你的Flink任务设了12个并行度，那Topic分区数也设12，这样每个并行度消费一个分区，效率最高。

数据采集通道搭建：端到端链路

数据采集通道，就是把模拟数据从生成器送到Kafka，再送到后续的实时处理层。我画了一张架构图，你看一眼就明白了。

整个链路跑起来后，你会看到数据从模拟器流出，经过Kafka，被Flink消费。这里有个关键点——一定要做数据校验。我习惯在Kafka消费者端加一个简单的校验逻辑，检查每条数据的字段是否完整、格式是否正确。

我的小技巧：在模拟器里故意混入一些「脏数据」，比如缺失字段、时间戳格式错误等。这样能提前验证你的清洗逻辑是否健壮。别问我为什么知道——我曾经上线第一天就被脏数据搞崩了实时链路。

采集通道的监控与告警

通道搭好了，不代表万事大吉。你需要关注几个核心指标：

数据延迟：从模拟器发出到Kafka落盘，延迟应该控制在毫秒级
消息堆积：Kafka消费者Lag不能持续增长
吞吐量：每秒处理的消息数，跟你的业务峰值匹配

我推荐用Prometheus + Grafana来监控这些指标。Kafka本身就暴露了JMX指标，你只需要配置一下Exporter就行。