3、数据采集层：用户行为日志采集（埋点）、消息队列选型（Kafka）、数据格式与序列化（Avro/Protobuf）
数据采集层，说白了就是推荐系统的「眼睛」和「耳朵」。
用户点了什么、看了多久、滑到了哪里、最终有没有下单——这些原始行为数据，全得靠这一层老老实实地收上来。我见过不少团队，算法模型搞得花里胡哨，结果底层数据不准，最后全白搭。所以这一章，咱们把埋点、消息队列、序列化这三块掰开揉碎讲清楚。
3.1 用户行为日志采集：埋点设计
埋点，就是你在代码里「钉钉子」——在用户发生关键行为的地方，把事件信息记录下来。我个人习惯把埋点分成三类：
| 类型 | 触发方式 | 典型场景 | 数据量级 |
|---|---|---|---|
| 曝光埋点 | 元素进入可视区域 | 商品卡片展示、推荐位曝光 | 极高（百万级/分钟） |
| 交互埋点 | 用户主动操作 | 点击、收藏、加购、分享 | 中（万级/分钟） |
| 业务埋点 | 后端服务回调 | 下单成功、支付完成、退款 | 低（千级/分钟） |
这里有个坑，我踩过不止一次——曝光埋点千万别在服务端做。为什么？因为服务端根本不知道用户「真正看到了什么」。你返回了10个商品，用户可能只看了前3个就划走了。曝光埋点必须在客户端，用IntersectionObserver或者类似机制，等元素真正进入视口再上报。
埋点数据长什么样？我一般用统一的事件模型：
{
"event_id": "uuid-xxxx", // 全局唯一ID，用于去重
"event_type": "item_click", // 事件类型
"timestamp": 1712345678000, // 客户端时间戳（毫秒）
"user_id": "u_10086", // 用户标识
"device_id": "d_abc123", // 设备标识
"page_id": "homepage_recommend", // 页面标识
"item_id": "sku_9527", // 目标对象ID
"extra": { // 扩展字段，按需填充
"position": 3, // 在推荐列表中的位置
"duration_ms": 2500, // 停留时长
"source": "personalized" // 推荐来源
}
}嗯，这里要注意：extra字段一定要设计成Map结构，别把业务字段写死在schema里。推荐场景的业务属性变化太快，今天加个「是否秒杀」，明天加个「直播间ID」，你总不能天天改协议吧？
3.2 消息队列选型：为什么是Kafka？
埋点数据收上来了，直接写数据库？别闹。推荐系统的数据量，峰值每秒几十万甚至上百万条，数据库根本扛不住。这时候就需要一个「缓冲池」——消息队列。
市面上消息队列不少，RabbitMQ、RocketMQ、Pulsar、Kafka……我为什么选Kafka？
核心原因就三个：
- 吞吐量天花板高：顺序读写磁盘，单机轻松扛10万+/秒
- 数据持久化：消息落盘，不怕丢数据。我习惯保留7天，方便回溯
- 消费位移管理：消费者挂了重启，能从断点继续消费，不丢不重
你想想看，推荐场景里最怕什么？最怕数据丢失。用户点了一个商品，结果没记录下来，那模型就永远学不到这个正样本。Kafka的ACK机制和ISR副本同步，能保证数据不丢——当然，前提是你配置对了。
我一般这样配置生产环境：
# 关键参数配置
acks=all # 所有副本确认才算成功
min.insync.replicas=2 # 至少2个副本同步
retention.ms=604800000 # 保留7天
compression.type=snappy # 启用压缩，减少网络带宽
batch.size=65536 # 64KB批量发送
linger.ms=10 # 延迟10ms攒批linger.ms设到10-20ms，让Kafka攒一批再发。这样吞吐量能提升3-5倍，CPU开销反而更低。
Topic怎么设计？我习惯按业务域拆分：
user_behavior_exposure— 曝光事件user_behavior_click— 点击事件user_behavior_conversion— 转化事件（加购、下单）user_behavior_custom— 自定义事件（AB实验、特殊标记）
分区数怎么定？一个经验公式：分区数 = max(生产者并发数, 消费者并发数)。我一般按消费者线程数的2倍来设，留点余量。比如消费者线程是8个，分区就设16个。
3.3 数据格式与序列化：Avro还是Protobuf？
数据进了Kafka，用什么格式存？JSON？别，千万别。JSON的解析开销大、体积大、没有强类型约束。生产环境里，我从来不用JSON做内部数据传输。
主流选择就两个：Avro 和 Protobuf。我两个都用过，说说我的感受。
| 对比维度 | Avro | Protobuf |
|---|---|---|
| Schema定义 | JSON格式，可读性强 | .proto文件，需要编译 |
| 序列化性能 | 中等，但无需生成代码 | 高，需要预编译 |
| Schema演进 | 天生支持，兼容性好 | 需要手动管理字段编号 |
| Kafka集成 | 原生支持，Confluent Schema Registry | 需要额外适配 |
| 数据体积 | 小（无字段名） | 极小（变长编码） |
我个人习惯：推荐系统内部用Avro。为什么？因为推荐场景的Schema变化太频繁了。今天加个字段，明天改个类型，Avro的Schema Registry能自动处理兼容性，不用改代码。Protobuf虽然性能更好，但每次改字段都要重新编译、发版，太折腾。
来看一个Avro Schema的例子：
{
"namespace": "com.recommend.schema",
"type": "record",
"name": "UserBehaviorEvent",
"fields": [
{"name": "event_id", "type": "string"},
{"name": "event_type", "type": "string"},
{"name": "timestamp", "type": "long"},
{"name": "user_id", "type": "string"},
{"name": "device_id", "type": "string"},
{"name": "page_id", "type": "string"},
{"name": "item_id", "type": "string"},
{"name": "extra", "type": {"type": "map", "values": "string"}, "default": {}},
{"name": "server_time", "type": ["null", "long"], "default": null}
]
}注意看server_time字段——我用了["null", "long"]的联合类型，默认值是null。这是Avro Schema演进的标准做法：新加的字段必须要有默认值，否则老数据反序列化会报错。
experiment_id字段，忘了设默认值。结果老数据全部反序列化失败，实时特征管道断了整整40分钟。从那以后，我定了个铁律：所有新增字段必须给默认值，且必须经过兼容性检查。
序列化之后的数据，配合Confluent Schema Registry使用，效果最佳。生产者只发数据的二进制内容，Schema ID放在前4个字节。消费者拿到数据后，根据Schema ID去Registry拉取对应的Schema，自动反序列化。这样，改Schema不用重启任何服务——只要保证向前兼容就行。
3.4 整体架构：一张图说清楚
讲了这么多，咱们用一张图把数据采集层的全貌串起来：
这张图里，我特别想强调一个点：Schema Registry不是可选项，是必需品。没有它，Avro的Schema演进优势就发挥不出来。你想想看，几十个微服务、几百个消费者，如果每次改Schema都要手动同步，那运维成本得有多高？
3.5 避坑指南：数据采集层的三个常见问题
最后，分享几个我实战中踩过的坑，希望能帮你少走弯路。
用户手机时间可能不准，甚至被恶意篡改。我的做法是：客户端和服务端各记一个时间戳。客户端时间用于计算「用户停留时长」这类相对值，服务端时间用于数据排序和窗口计算。两个时间戳差值超过5分钟的，直接标记为异常数据。
有次大促，流量突增10倍，消费者来不及消费，消息积压了上亿条。后来我加了动态扩缩容机制：监控消费延迟，超过阈值自动增加消费者线程数。同时，Kafka的
max.poll.records不要设太大，我一般控制在500-1000，防止单次拉取太多导致处理超时。
前面提过，新增字段必须给默认值。还有一个容易忽略的点：不要删除字段。如果你觉得某个字段没用了，标记为deprecated就行，别删。删了字段，老数据反序列化直接报错。实在要删，等所有消费者都升级完了再动手。
数据采集层，看起来就是「收数据、传数据、存数据」三个步骤，但每个环节都有不少细节。埋点不准，后面全白干；消息队列选错，性能瓶颈卡死你；序列化没选好，改个字段要发版三天。把这些基础打扎实了，上层才能盖得稳。