3、数据源与数据采集：用户行为日志设计、埋点规范、Kafka生产者与消费者实战

数据采集是整个实时推荐平台的起点。说白了，没有高质量的用户行为数据，后面再牛的算法也是白搭。我见过不少团队，模型调得飞起，结果上线后发现数据质量一塌糊涂——字段缺失、时间戳错乱、重复消费……嗯，最后全得回头补数据基建的课。

这一章我们就来聊聊，怎么把用户行为日志这件事做扎实。从埋点规范到 Kafka 生产消费，一条线串下来。

3.1 用户行为日志：到底要记什么

用户行为日志，说白了就是记录「谁在什么时间、什么地方、做了什么事」。但具体到字段设计，很多新手容易漏掉关键信息。

我个人习惯把日志分为三类：

• 曝光日志：用户看到了什么（商品、广告、推荐位）
• 交互日志：用户做了什么（点击、滑动、收藏、加购）
• 转化日志：用户最终完成了什么（下单、支付、注册）

这三类日志的优先级不一样。曝光日志量最大，但信噪比低；转化日志量最小，但价值最高。我在项目中遇到过一个问题：曝光日志一天几十亿条，但真正被点击的不到 1%。后来我们加了采样策略，曝光日志只保留 10%，交互和转化全量保留——效果立竿见影。

3.2 埋点规范：别让数据变成垃圾

埋点不规范，数据两行泪。这不是开玩笑。我见过一个团队，同一个「商品点击」事件，iOS 叫 click_item，Android 叫 item_click，Web 叫 productClick——最后做数据聚合时，光字段映射就写了 200 行代码。

所以，埋点规范必须在项目启动时就定死。我推荐以下这套命名规则：

维度	规范	示例
事件命名	小写 + 下划线，动词_名词	click_item、view_page
属性命名	小驼峰，不超过 3 级	itemId、pageName
时间戳	毫秒级 Unix 时间戳	1710000000000
用户标识	统一使用 deviceId + userId	deviceId: "abc123"

3.3 日志结构设计：JSON 是王道

日志格式我推荐用 JSON，原因很简单——可读性强、扩展性好、解析方便。但 JSON 也有坑：字段太多会导致存储膨胀，解析性能下降。

我常用的日志结构长这样：

{
  "event": "click_item",
  "timestamp": 1710000000000,
  "deviceId": "abc123",
  "userId": "user_456",
  "properties": {
    "itemId": "item_789",
    "pageName": "recommend_feed",
    "position": 3,
    "duration": 2500
  },
  "context": {
    "os": "Android 14",
    "appVersion": "3.2.1",
    "network": "wifi"
  }
}

注意几个设计要点：

• event 和 timestamp 放在最外层，方便快速过滤
• properties 放业务属性，按事件类型动态变化
• context 放环境信息，所有事件共用一套

为什么这么分？我在项目中踩过坑：一开始把所有字段平铺，结果加一个新属性就要改所有下游的解析逻辑。后来改成嵌套结构，新增属性只需要在 properties 里加字段，下游按需取用，互不影响。

3.4 Kafka 生产者实战：把日志送进管道

日志采集好了，怎么送到后端？Kafka 是目前最主流的选择。它的核心优势就两个：高吞吐、持久化。

先看生产者代码。我用 Java 写一个简单的示例：

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");
// 开启幂等性，防止重复
props.put("enable.idempotence", true);
// 设置 acks=all，保证数据不丢
props.put("acks", "all");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

// 发送日志
String logJson = "{\"event\":\"click_item\",\"timestamp\":1710000000000}";
ProducerRecord<String, String> record = new ProducerRecord<>("user_behavior_log", logJson);
producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        // 记录失败日志，后续重试
        log.error("发送失败", exception);
    }
});

这里有几个关键配置，我重点说一下：

• enable.idempotence=true：开启幂等性，防止网络抖动导致消息重复。我早期没开这个，结果下游消费端收到大量重复数据，推荐结果直接乱套。
• acks=all：等待所有副本确认，保证消息不丢。吞吐量会降一点，但数据可靠性优先。
• 回调处理：异步发送一定要加回调，否则发送失败你完全不知道。

3.5 Kafka 消费者实战：把数据吃进来

生产者把日志送进 Kafka，消费者负责把数据拉出来处理。实时推荐场景下，消费者通常是 Flink 或 Spark Streaming，但为了讲清楚原理，我们先看原生消费者：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "recommend_consumer_group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
// 自动提交偏移量，建议关闭
props.put("enable.auto.commit", "false");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("user_behavior_log"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        // 解析日志
        String log = record.value();
        // 处理业务逻辑
        processLog(log);
    }
    // 手动提交偏移量
    consumer.commitSync();
}

消费者设计有几个要点：

• 关闭自动提交：enable.auto.commit=false，改为手动提交。否则处理失败时偏移量已经提交，数据就丢了。
• 消费组设计：不同业务用不同的 group.id。比如推荐系统用 recommend_group，数据分析用 analytics_group，互不影响。
• 批量处理：poll() 一次拉一批数据，不要一条一条处理，否则吞吐量上不去。

3.6 避坑指南：数据采集的五个常见问题

做数据采集这些年，我踩过的坑能写一本书。挑几个最常见的分享给你：

1. 时间戳不一致：客户端时间可能不准，建议服务端收到日志后加盖一个 serverTimestamp，以服务端时间为准。
2. 消息重复消费：幂等性 + 手动提交偏移量 + 消费端去重，三层防护。
3. Topic 分区倾斜：如果某个用户行为特别多（比如大促时的爆款商品），会导致某个分区数据量暴增。建议按 userId 哈希分区，而不是按 event 类型。
4. 序列化性能：JSON 虽然方便，但解析慢。高吞吐场景可以考虑 Avro 或 Protobuf，压缩率高、解析快。
5. 日志积压告警：监控 Kafka 消费延迟，超过阈值立刻告警。我见过一个团队，消费延迟了 2 小时才发现，推荐结果全是过时的。