3、实时数据采集与接入：Kafka 核心概念、生产者/消费者 API、数据入湖入仓策略

聊到实时数据产品，Kafka 几乎是绕不开的基石。我见过不少团队，一上来就追求复杂的 Flink 逻辑，结果发现数据源头就歪了——要么丢消息，要么重复消费，要么入湖入仓后对不上账。说白了，实时数据产品化的第一步，就是把 Kafka 这层「管道」玩明白。

这一章，我会从核心概念讲起，结合生产者和消费者 API 的实战经验，再聊聊数据入湖入仓的那些坑。嗯，都是我在项目中摔过的跟头。

3.1 Kafka 核心概念：不只是消息队列

很多人把 Kafka 当成普通消息队列用，其实它更像一个「分布式日志系统」。我个人习惯把它的核心抽象成四个词：Topic、Partition、Offset、Consumer Group。

• Topic：数据的逻辑分类，好比数据库里的表。
• Partition：Topic 的物理分片，是并行度的来源。分区数决定了消费的最大并发度。
• Offset：消息在分区内的唯一序号。消费进度全靠它维护。
• Consumer Group：一组消费者的逻辑集合。组内每个分区只能被一个消费者消费，这是实现负载均衡和容错的关键。

我曾经帮一个金融客户排查数据乱序问题，他们用了 8 个分区，结果下游做聚合时发现时间戳倒挂。查了半天，原来是同一个用户 ID 的消息被路由到了不同分区。解决方案很简单：用用户 ID 作为消息 key，确保相同 key 进入同一分区。

3.2 生产者 API：如何把数据「喂」给 Kafka

生产者 API 看起来简单——就一个 send() 方法。但实际用起来，细节多到让人头疼。

3.2.1 核心参数调优

参数	默认值	我的建议
acks	1	对数据不敏感用 1，金融场景用 all（配合 min.insync.replicas）
linger.ms	0	吞吐优先设为 5-20ms，延迟优先保持 0
batch.size	16384	根据消息大小调整，通常 32KB-64KB 比较均衡
compression.type	none	文本类数据强烈推荐 snappy 或 lz4，能省 50% 带宽
retries	2147483647	配合 enable.idempotence=true 使用，避免重复

3.2.2 发送模式选择

Kafka 生产者支持三种发送模式，你想想看，选错了会有什么后果？

1. 发后即忘（Fire-and-Forget）：不关心结果，吞吐最高，但可能丢消息。适合日志采集等可容忍少量丢失的场景。
2. 同步发送：调用 get() 等待结果，吞吐最低。我一般只在初始化或测试时用。
3. 异步回调：通过 Callback 处理结果，兼顾吞吐和可靠性。这是生产环境的首选。

// 异步回调示例
producer.send(new ProducerRecord<>("order-events", key, value),
    (metadata, exception) -> {
        if (exception != null) {
            // 记录失败消息到本地文件或死信队列
            log.error("发送失败，key: {}", key, exception);
        }
    });

3.3 消费者 API：如何「接住」数据

消费者 API 的难点不在消费，而在「如何正确地消费」。我见过太多因为消费位移管理不当导致的数据重复或丢失。

3.3.1 位移提交策略

策略	说明	适用场景
自动提交（enable.auto.commit=true）	每隔 auto.commit.interval.ms 自动提交	对重复不敏感，处理逻辑幂等
手动同步提交	处理完一批消息后调用 commitSync()	需要精确一次语义，吞吐要求不高
手动异步提交	调用 commitAsync() 配合回调	吞吐优先，能接受少量重复

3.3.2 再均衡监听器

消费者组内成员变化时，会触发再均衡（Rebalance）。如果不处理，正在消费的分区会被分配给其他消费者，导致数据重复或丢失。

// 再均衡监听器示例
consumer.subscribe(Collections.singletonList("order-events"),
    new ConsumerRebalanceListener() {
        @Override
        public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
            // 在分区被收回前，提交当前位移
            consumer.commitSync(currentOffsets);
        }

        @Override
        public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
            // 可以在这里做状态初始化
            log.info("新分配分区: {}", partitions);
        }
    });

3.4 数据入湖入仓策略

数据从 Kafka 出来，最终要落到数据湖或数据仓库里。这一步看似简单，实则暗流涌动。

3.4.1 入湖 vs 入仓：选哪个？

维度	数据湖（如 Hudi/Iceberg）	数据仓库（如 ClickHouse/Doris）
数据格式	原始格式，支持 Schema Evolution	结构化，Schema 固定
写入频率	微批或流式，分钟级常见	秒级甚至毫秒级
查询延迟	分钟级（批查询）	秒级（OLAP 查询）
典型场景	历史回溯、机器学习特征	实时报表、大屏、Ad-hoc 查询

我的经验是：先入湖，再入仓。数据先以原始格式落到湖里，保证不丢不重；然后通过流式 ETL 清洗后写入数仓。这样既保留了原始数据的可回溯性，又满足了实时查询的需求。

3.4.2 入湖策略：选对文件格式

数据湖的核心是文件格式。目前主流的是 Apache Hudi、Apache Iceberg 和 Delta Lake。我个人更倾向 Iceberg，原因有三：

• Schema Evolution 更灵活：支持增删列、重命名列，不用重写历史数据。
• 隐藏分区：不用手动管理分区目录，Iceberg 自动处理。
• Time Travel：可以查询任意时间点的数据快照，排查问题利器。

3.4.3 入仓策略：实时数仓分层

实时数仓通常也遵循分层思想，但比离线数仓更「薄」：

1. ODS（操作数据层）：直接从 Kafka 消费，不做任何加工，保留原始数据。入湖后的数据就是 ODS。
2. DWD（明细数据层）：做数据清洗、字段补全、维度关联。比如把用户 ID 关联出用户所在城市。
3. DWS（汇总数据层）：按业务维度做预聚合。比如每分钟统计各城市的订单量。
4. ADS（应用数据层）：面向具体报表或大屏，数据量最小，查询最快。

-- DWS 层示例：每分钟订单汇总
INSERT INTO dws_order_minute
SELECT
    window_start,
    city_id,
    COUNT(DISTINCT order_id) AS order_cnt,
    SUM(amount) AS total_amount
FROM dwd_order_detail
GROUP BY
    TUMBLE(event_time, INTERVAL '1' MINUTE),
    city_id;

3.5 本章小结：一张图看懂数据流转

说了这么多，不如一张图来得直观。下面这张 SVG 图展示了从数据采集到入湖入仓的完整链路：

这张图把整个链路串起来了。你想想看，从数据源到 Kafka，再到消费者，最后分流到数据湖和数据仓库，每一步都有对应的策略和参数。我在实际项目中，通常会在消费者层做一次「数据分流」：一份入湖做历史回溯，一份入仓做实时查询，互不干扰。