3. 数据源接入：常见实时数据源对比与选型

实时特征生产的第一步是稳定、低延迟地接入数据源。消息队列作为数据管道核心组件，其选型直接决定链路的吞吐、延迟与运维复杂度。本章从主流消息队列对比、选型原则、数据格式三个维度展开。

3.1 主流实时数据源对比：Kafka vs. Kinesis vs. Pulsar

当前工业界最常用的实时数据源为 Apache Kafka、Amazon Kinesis 和 Apache Pulsar。三者均支持发布-订阅模型，但在架构、性能、运维上存在显著差异。

3.2 消息队列选型原则

在实时特征生产场景下，消息队列选型需遵循以下五项核心原则：

1. 延迟优先： 特征生产对端到端延迟敏感（通常要求 < 100ms），优先选择毫秒级延迟的消息队列。
2. 吞吐与扩展性： 根据峰值流量预估分区/Shard 数，确保队列可水平扩展，避免单点瓶颈。
3. 消息可靠性： 至少一次语义（At-Least-Once）是底线；关键特征链路需支持 Exactly-Once 语义。
4. 重播与回溯： 模型训练需要历史数据回溯，队列必须支持按时间或 Offset 重播。
5. 运维成本： 自建 vs. 托管需权衡。小团队优先托管服务，大团队可自建以降低成本。

3.3 数据格式选择：Avro vs. Protobuf vs. JSON

消息队列中传输的数据格式直接影响序列化/反序列化性能、Schema 演进能力与可读性。三种主流格式对比如下：

3.4 数据格式选型决策流程

下图展示了在实时特征生产场景下，如何根据业务需求选择数据格式：

3.5 数据格式与消息队列的集成示例

以下是一个使用 Kafka + Avro + Schema Registry 的生产者/消费者代码片段，展示实时特征数据的序列化与反序列化：

// 生产者：使用 Avro 序列化特征数据
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
props.put("schema.registry.url", "http://localhost:8081");

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

FeatureRecord feature = FeatureRecord.newBuilder()
    .setUserId("user_12345")
    .setFeatureName("click_rate_7d")
    .setFeatureValue(0.8765)
    .setTimestamp(System.currentTimeMillis())
    .build();

ProducerRecord<String, FeatureRecord> record =
    new ProducerRecord<>("feature-input", feature.getUserId(), feature);
producer.send(record);
producer.close();

// 消费者：使用 Avro 反序列化特征数据
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "feature-consumer-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "io.confluent.kafka.serializers.KafkaAvroDeserializer");
props.put("schema.registry.url", "http://localhost:8081");
props.put("specific.avro.reader", "true");

KafkaConsumer<String, FeatureRecord> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("feature-input"));

while (true) {
    ConsumerRecords<String, FeatureRecord> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, FeatureRecord> record : records) {
        FeatureRecord feature = record.value();
        System.out.printf("User: %s, Feature: %s, Value: %f%n",
            feature.getUserId(), feature.getFeatureName(), feature.getFeatureValue());
    }
}

3.6 本章小结

• 消息队列选型： Kafka 适合自建高吞吐场景，Kinesis 适合 AWS 全托管，Pulsar 适合多租户与跨地域复制。
• 数据格式选型： 实时特征生产优先选择 Avro + Schema Registry，兼顾性能与 Schema 演进；Protobuf 适合 gRPC 服务；JSON 仅用于调试。
• 集成要点： 生产者与消费者必须使用相同的 Schema Registry，并配置兼容性策略，避免运行时反序列化失败。