实时数据源接入：Kafka基础概念与架构、Kafka生产者与消费者API、Flink Kafka Connector实战

好，咱们直接进入正题。实时特征工程的第一步，说白了就是「怎么把数据实时拿过来」。你可能会想，用HTTP轮询行不行？行，但延迟高、资源浪费严重。生产环境里，我几乎没见过谁用轮询做实时特征。大家用的都是消息队列，而Kafka，就是消息队列里的「事实标准」。

我最早接触Kafka是在2016年，那时候公司要做用户实时行为画像。数据源是App的点击流，每秒几万条。用RabbitMQ试过，吞吐量上不去，消费端还经常堆积。后来换成Kafka，问题一下就解决了。嗯，从那以后，我对Kafka的定位就很清晰——它是为「高吞吐、低延迟、持久化」而生的。

Kafka核心概念与架构

先看一张整体架构图，帮你建立直观印象。

这张图里，Producer往Kafka集群里写数据，Consumer从集群里读数据。中间的Broker就是Kafka服务器，Topic是数据的分类。你想想看，如果每秒要处理几十万条用户点击，没有这种架构根本扛不住。

我个人习惯把Kafka理解成一个「超级缓冲区」。它不负责计算，只负责把数据存下来、按顺序分发。为什么它快？因为它是顺序写磁盘，而不是随机写。机械硬盘顺序写能到200MB/s，比随机写快两个数量级。

Kafka生产者与消费者API

光懂架构不够，得能写代码。咱们先看生产者API。说白了，就是往Kafka里发消息。

// 生产者核心配置
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=all 保证数据不丢
props.put("acks", "all");

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

// 发送消息
ProducerRecord<String, String> record = 
    new ProducerRecord<>("user_click", "user_1001", "{\"event\":\"click\",\"item_id\":\"123\"}");
producer.send(record, (metadata, exception) -> {
    if (exception == null) {
        System.out.println("发送成功，偏移量：" + metadata.offset());
    } else {
        System.err.println("发送失败：" + exception.getMessage());
    }
});
producer.close();

这里有个坑，我踩过。刚开始用Kafka时，我把acks设成0，觉得这样快。结果线上丢数据了，用户行为少了一大截。后来改成all，虽然延迟多了几毫秒，但数据一条不丢。实时特征工程里，数据完整性比那几毫秒延迟重要得多。

再看消费者API。消费者从Kafka拉数据，然后处理。

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "feature-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_click"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        // 这里就是特征计算的入口
        System.out.printf("offset = %d, key = %s, value = %s%n", 
            record.offset(), record.key(), record.value());
    }
    // 手动提交，确保处理完再提交偏移量
    consumer.commitSync();
}

Flink Kafka Connector实战

好了，现在数据在Kafka里，怎么把它喂给Flink做特征计算？Flink官方提供了Kafka Connector，用起来非常顺手。

先加依赖：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka_2.12</artifactId>
    <version>1.17.0</version>
</dependency>

然后写Flink作业，从Kafka读数据：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4);

// 配置Kafka消费者
Properties kafkaProps = new Properties();
kafkaProps.setProperty("bootstrap.servers", "localhost:9092");
kafkaProps.setProperty("group.id", "flink-feature-group");

DataStream<String> stream = env.addSource(
    new FlinkKafkaConsumer<>(
        "user_click",
        new SimpleStringSchema(),
        kafkaProps
    )
);

// 实时特征计算：统计每个用户过去5分钟的点击次数
stream
    .map(value -> {
        JSONObject json = new JSONObject(value);
        return Tuple2.of(json.getString("user_id"), 1L);
    })
    .keyBy(value -> value.f0)
    .window(TumblingProcessingTimeWindows.of(Time.minutes(5)))
    .sum(1)
    .print();

env.execute("Real-time Feature Pipeline");

这段代码做了什么？它从Kafka的user_click主题里读数据，然后按用户ID分组，每5分钟统计一次点击次数。这就是一个最简单的实时特征。

你可能会问，为什么不用Spark Streaming？嗯，Flink在实时场景下延迟更低，毫秒级。Spark Streaming本质上是微批处理，延迟至少几百毫秒。做实时特征，延迟就是金钱。我做过对比，同样一个滑动窗口特征，Flink延迟比Spark Streaming低一个数量级。

最后，给你一个实战中的配置清单，照着做基本不会出问题：

配置项	推荐值	说明
bootstrap.servers	至少3个Broker地址	避免单点故障
acks	all	保证数据不丢
enable.auto.commit	false	手动提交，精确控制
Flink Checkpoint间隔	30秒	平衡恢复时间和性能
Topic分区数	≥ Flink并行度	充分利用并行能力

好了，这一章的内容就到这里。Kafka作为实时数据源的核心，你只要把架构、API和Flink Connector这三块吃透，后面的特征计算就水到渠成了。