一条行为日志要进Kafka又要进HDFS——怎么避免重复采集和数据不一致？

数据采集手册第 7 题约 10 分钟读完

本文目录

先说结论
业务场景：实时离线两套链路打架
方案一：Kafka作为单一事实源（推荐架构）
方案二：Flink流式落HDFS的实现细节
方案三：Kafka Connect HDFS Sink（开箱即用）
方案对比
面试高频追问
实战踩坑：Flink落HDFS的小文件雪崩
总结

先说结论

正确架构原则：一次采集、多端消费。日志只采一份到Kafka，实时从Kafka直接消费，离线从Kafka落HDFS再批处理。

一致性保证靠：Kafka作为唯一事实来源（Single Source of Truth），实时离线都从同一个offset消费，不存在两份数据。

常见误区：用两套Agent分别采（重复+不一致）；离线直接读日志文件不经Kafka（漂移+回放难）。

先把这段讲顺；面试官追问时，再展开下面的原理和细节。

详细讲解

"实时链路要、离线链路也要"是大数据架构里最常见的诉求。直觉上可能想让两条链路各采一次，但这样马上就出现两个问题：日志服务器双倍IO，实时和离线还可能数据不一致。这道题面试官想看的是你有没有Lambda/Kappa架构的概念，能不能设计出"一次采集、多端消费"的方案。

业务场景：实时离线两套链路打架

一个用户行为日志架构最初的设计是这样的：实时分析用Flume+Kafka+Flink，离线分析用Logstash+ES（顺便给离线分析），结果业务方发现"昨天的UV"实时大盘和T+1报表对不上，差异3-5%。复盘后才发现，两套Agent对日志的解析逻辑、丢失率、时间窗口都不一样，根本不可能完全一致。

❌ 反模式：双Agent各采各的
 
  日志文件 ──┬─ Flume Agent  → Kafka  → Flink (实时大盘)
             └─ Logstash    → ES     → 离线分析
 
  问题：
  - 两个Agent解析规则容易不一致
  - 各自的丢失率不同(网络抖动时差异更大)
  - 时间字段处理可能不同(本地时间vs UTC)
  - 日志服务器双倍IO开销

方案一：Kafka作为单一事实源（推荐架构）

最佳实践是"一次采集、多次消费"。日志只通过一条链路进Kafka，实时和离线都从这个Kafka消费，本质上读的是同一份数据。

✓ 正确架构：Kafka做hub
 
                       ┌─→ Flink (实时大盘)
  日志文件 ─→ Filebeat ─→ Kafka ─┼─→ Spark SS (实时数仓)
                       │       └─→ HDFS Sink (落HDFS分区,Spark批处理)
                       │           ↓
                       │      离线数仓DWD/DWS
                       │
                       └─→ ES Sink (检索)
 
  优势：
  - 数据来源唯一，实时离线天然一致
  - 日志服务器只采集一次，资源开销最小
  - 新接入消费方零成本(再起一个Consumer Group)
  - Kafka保留3-7天，下游故障可以回放

具体落地有两种方式：① 用Kafka Connect HDFS Sink，原生支持exactly-once；② 用Flink流式写HDFS（StreamingFileSink），更灵活但要自己保证一致性。

方案二：Flink流式落HDFS的实现细节

用Flink把Kafka数据落到HDFS分区，是最灵活的方式。关键是用StreamingFileSink + Checkpoint保证Exactly-Once。

// Flink流式落HDFS(按小时分区)
StreamingFileSink<LogEvent> sink = StreamingFileSink
    .forRowFormat(new Path("hdfs:///data/ods/user_log"),
                  new SimpleStringEncoder<LogEvent>("UTF-8"))
    .withBucketAssigner(new DateTimeBucketAssigner<>(
        "yyyy-MM-dd/HH",  // 按小时分桶
        ZoneId.of("Asia/Shanghai")))
    .withRollingPolicy(
        DefaultRollingPolicy.builder()
            .withRolloverInterval(TimeUnit.MINUTES.toMillis(15))   // 15分钟切文件
            .withInactivityInterval(TimeUnit.MINUTES.toMillis(5))   // 5分钟无数据切
            .withMaxPartSize(128 * 1024 * 1024)                    // 128MB切
            .build())
    .withOutputFileConfig(OutputFileConfig.builder()
            .withPartPrefix("part").withPartSuffix(".parquet").build())
    .build();
 
// 必须开启Checkpoint，且interval ≤ rollover时间
env.enableCheckpointing(60_000, CheckpointingMode.EXACTLY_ONCE);
// HDFS文件三状态: in-progress -> pending -> finished(checkpoint触发)

关键点：StreamingFileSink只在Checkpoint成功后才把文件状态从pending改成finished。这就是Exactly-Once的根本——失败重启后，未finished的文件直接丢弃，从Kafka offset重新消费写入。

方案三：Kafka Connect HDFS Sink（开箱即用）

如果不想自己写Flink作业，Kafka Connect的HDFS Sink Connector能直接把topic数据落HDFS，配置即可，原生支持Exactly-Once。

{
  "name": "hdfs-user-log-sink",
  "config": {
    "connector.class": "io.confluent.connect.hdfs3.Hdfs3SinkConnector",
    "tasks.max": "4",
    "topics": "user_log",
    "hdfs.url": "hdfs://namenode:8020",
    "flush.size": "10000",                          // 1万条flush一次
    "rotate.interval.ms": "600000",                  // 10分钟轮转
    "path.format": "'dt'=YYYY-MM-dd/'hr'=HH",
    "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "timestamp.extractor": "RecordField",            // 用消息里的时间字段
    "timestamp.field": "event_ts",
    "format.class": "io.confluent.connect.hdfs.parquet.ParquetFormat",
    "hadoop.conf.dir": "/etc/hadoop/conf"
  }
}

适用场景：已有Kafka Connect集群；数据简单不需要做ETL；运维不熟Flink。

方案对比

维度	Flink流式落HDFS	Kafka Connect	双Agent(反模式)
一致性	Exactly-Once(Checkpoint)	Exactly-Once(原生)	无法保证
灵活性	高(可做ETL)	低(原样转储)	中
运维复杂度	中(Flink集群)	低(配置即可)	高(两套Agent)
资源开销	中	低	高(双采集)
最佳场景	需要ETL的实时数仓	只是落地存储	禁用

面试高频追问

追问一：Kafka里的数据要保留多久？

看下游故障容忍窗口。生产上一般保留7天。原因：① 离线作业最长可能跑6小时，加上故障排查时间，72小时内能接住；② 离线追数据场景（比如DWS层修复要重跑3天数据）；③ 7天能覆盖大部分故障窗口，14天太长存储成本高。如果业务对追数要求高，加个长期归档存储（OSS/S3）做冷备。

追问二：HDFS落地后小文件多怎么办？

两个方向：① 写入侧——StreamingFileSink的rolling policy配大一点（128MB或30分钟），Kafka Connect的flush.size调大；② 后处理——每天凌晨跑一个Spark合并作业，把当天的小文件合成128MB的大文件，删除原文件。我一般两个都做，写入侧先减少小文件产生，后处理兜底。

追问三：Kafka压缩topic vs 普通topic用哪种？

行为日志这种append-only数据用普通topic+时间过期。压缩topic（log compaction）适合"保留每个key最新值"的场景，比如user_profile表的binlog同步。误用压缩topic会导致历史日志被清掉，离线作业重跑时数据不全。这是新手常见错误。

实战踩坑：Flink落HDFS的小文件雪崩

第一次上Flink流式落HDFS时没认真看RollingPolicy，用了默认值（128MB或60秒），结果业务低峰期数据量小，1分钟就切一个文件，每天产生几万个小文件。HDFS NameNode内存压力暴增，Hive查询性能也变差。

// ❌ 默认配置：低峰期60秒就切，小文件爆炸
DefaultRollingPolicy.builder().build()  // 默认1分钟+128MB
 
// ✓ 优化配置：拉长时间窗口
DefaultRollingPolicy.builder()
    .withRolloverInterval(TimeUnit.MINUTES.toMillis(30))    // 改成30分钟
    .withInactivityInterval(TimeUnit.MINUTES.toMillis(15))   // 15分钟无数据才切
    .withMaxPartSize(256 * 1024 * 1024)                     // 256MB
    .build()
 
// 同时加每日合并作业
// 每天凌晨2点把昨天的所有小文件合成128MB一个的大文件
spark.read.parquet("/data/ods/user_log/dt=2024-01-15/")
     .repartition(50)  // 控制输出文件数
     .write.mode("overwrite").parquet("/data/ods/user_log/dt=2024-01-15/")

解决方案：写入侧调RollingPolicy降低小文件产生速度，每日合并作业兜底。HDFS小文件问题在所有流式落地场景都会出现，必须前置处理，等NameNode告警就晚了。

总结

"实时+离线"双链路的最佳架构是Kafka作为单一事实源，下游各自从Kafka消费，自然保证一致性。Flink流式落HDFS用Checkpoint机制做Exactly-Once，注意小文件问题。坚决避免双Agent反模式。面试时按"反模式→正模式→实现细节→踩坑"四步讲，体现架构思维分高。

下一篇聊一个Filebeat的实战题：日志写入太快Filebeat跟不上，怎么调优才不丢数据？

你们公司实时和离线对账经常对不上吗？怎么解决的？评论区聊聊～

关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长

刷题进度保存在当前浏览器