3、实时数据源接入：Kafka基础概念与架构、Kafka生产者与消费者API、Kafka主题与分区策略

好，咱们进入第三章。实时特征工程，数据从哪来？十有八九是从 Kafka 来。我见过不少团队，特征工程做得挺花哨，结果数据源没接稳，整个 pipeline 天天报警。说白了，Kafka 就是实时数据的「高速公路」。这一章，咱们就把这条路怎么修、怎么跑车，彻底搞清楚。

3.1 Kafka 基础概念与架构

先聊聊 Kafka 到底是什么。它本质上是一个分布式消息队列系统，但比传统 MQ 能扛的吞吐量高得多。我最早接触 Kafka 是在做用户点击流处理的时候，一天几十亿条消息，RabbitMQ 直接跪了，Kafka 稳如老狗。

核心角色

• Producer（生产者）：发消息的。比如你的应用服务把用户行为日志扔进 Kafka。
• Consumer（消费者）：收消息的。实时特征计算任务就是典型的消费者。
• Broker（代理节点）：存消息的服务器。一个 Kafka 集群由多个 Broker 组成。
• Topic（主题）：消息的分类。好比数据库里的表，同一类消息进同一个 Topic。
• Partition（分区）：Topic 的物理分片。这是 Kafka 能横向扩展的关键。
• Offset（偏移量）：消息在分区内的唯一编号。消费者靠它知道自己读到哪了。

架构长什么样？

我画了一张图，把 Kafka 的核心流转画出来了。你看一眼就明白。

3.2 Kafka 生产者与消费者 API

概念懂了，咱们上手写代码。实时特征工程里，你大概率既要写 Producer 把数据送进去，也要写 Consumer 把数据拉出来处理。我习惯用 Java 客户端，但 Python 的 confluent-kafka 也很流行。这里我用 Java 举例，因为生产环境里 Java 生态更成熟。

3.2.1 生产者 API

写 Producer 其实就三步：配参数、建客户端、发消息。嗯，这里有个坑——消息确认机制。

// 生产者核心配置
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");
// 重点：acks 参数
props.put("acks", "all");  // 等待所有副本确认，最安全但最慢

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

// 发送消息
ProducerRecord<String, String> record = 
    new ProducerRecord<>("user_click", "user_123", "{\"page\":\"/home\",\"ts\":1700000000}");

producer.send(record, (metadata, exception) -> {
    if (exception == null) {
        System.out.println("发送成功，分区：" + metadata.partition() + 
                           "，偏移量：" + metadata.offset());
    } else {
        System.err.println("发送失败：" + exception.getMessage());
    }
});

producer.close();

3.2.2 消费者 API

消费者这边，核心是「拉取模型」。Kafka 不会主动推数据给你，你得自己去拉。还有一个关键概念叫「消费者组」——同一个组内的消费者共同消费一个 Topic，每条消息只被组内一个消费者处理。

// 消费者核心配置
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "feature-compute-group");  // 消费者组
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
// 重点：偏移量重置策略
props.put("auto.offset.reset", "earliest");  // 从最早的消息开始消费

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

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        // 这里就是实时特征计算的入口
        System.out.printf("分区=%d, 偏移量=%d, 键=%s, 值=%s%n",
            record.partition(), record.offset(), record.key(), record.value());
        
        // 实际项目中：解析 JSON，提取特征，更新状态
        processFeature(record.value());
    }
    // 手动提交偏移量，防止重复消费
    consumer.commitSync();
}

3.3 Kafka 主题与分区策略

Topic 和 Partition 的设计，直接决定了你的实时特征系统能扛多大流量。我见过不少团队，Topic 建了，分区数随便设个 3，结果流量一上来，消费者处理不过来，消息积压到天荒地老。

分区数怎么定？

没有标准答案，但我有个经验公式：

因素	建议
目标吞吐量	单分区约 10MB/s，按峰值流量反推
消费者并行度	分区数 ≥ 消费者数，否则有消费者闲置
顺序性要求	需要严格顺序的场景，用单分区或按 key 路由
Broker 数量	分区数最好是 Broker 数的整数倍，负载均衡

分区策略：消息到底进哪个分区？

Producer 发消息时，可以指定分区策略。默认策略是轮询（Round-Robin），消息均匀分布。但实时特征场景下，我经常用「按 key 分区」——同一个用户的消息进同一个分区，这样消费者可以按用户维度聚合特征。

// 自定义分区器：按用户 ID 哈希分区
props.put("partitioner.class", "com.example.UserIdPartitioner");

// 或者更简单：在 ProducerRecord 里指定分区
ProducerRecord<String, String> record = 
    new ProducerRecord<>("user_click", 
                         Math.abs(userId.hashCode()) % partitionCount,  // 指定分区
                         userId, 
                         jsonValue);

分区与消费者的关系

一个分区只能被同一个消费者组内的一个消费者消费。但一个消费者可以消费多个分区。这就是 Kafka 实现并行消费的机制。

3.4 实时特征场景下的最佳实践

最后，我把这些年做实时特征工程时，关于 Kafka 接入的一些经验总结一下。你直接拿去用就行。

1. Topic 命名规范：用 环境_数据域_业务 格式，比如 prod_user_click、test_order_pay。别乱起名，不然运维想骂人。
2. 消息体设计：用 JSON 或 Avro，别用纯文本。JSON 可读性好，Avro 压缩率高。我一般先用 JSON 快速迭代，稳定后切 Avro。
3. 消息大小控制：单条消息别超过 1MB。Kafka 默认限制就是 1MB，超了会报错。特征数据一般很小，但如果你塞了大图片或大文本，记得拆分。
4. 监控与告警：必须监控消费者 Lag（滞后量）。Lag 持续增长说明消费速度跟不上生产速度，要么加分区，要么优化消费逻辑。
5. 幂等生产者：开启 enable.idempotence=true，防止网络抖动导致消息重复。实时特征场景下，重复消息会污染特征值。