场景背景：数据时间错位问题

在典型的双层Flume架构中，第一层通过TailDir Source采集日志并写入Kafka Channel，第二层由Kafka Source消费数据后落地HDFS。此架构存在一个隐性风险：日志产生时间与数据到达HDFS的时间可能存在跨天偏差。

例如，23:59:59产生的日志经过Kafka缓冲，到达第二层Flume时可能已是次日00:00:XX。若按系统时间分区，该条记录将被错误地归档到次日目录，造成数据统计口径不一致。

HDFS Sink日期分区机制

Flume的HDFS Sink支持时间占位符解析，配置示例如下：

a1.sinks.hdfsSink.type = hdfs
a1.sinks.hdfsSink.hdfs.path = /warehouse/log/raw/%Y-%m-%d
a1.sinks.hdfsSink.hdfs.filePrefix = events-
a1.sinks.hdfsSink.hdfs.round = false

其中%Y-%m-%d的解析依赖于Event Header中的timestamp字段。Kafka Source默认以当前系统时间为该字段赋值，这正是导致数据漂移的根本原因。

自定义拦截器实现方案

通过自定义Interceptor，从日志体中提取业务时间戳，覆盖Header中的默认值：

package com.example.flume.interceptor;

import com.google.gson.Gson;
import com.google.gson.JsonObject;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

public class LogTimeExtractor implements Interceptor {
    
    private final Gson gson = new Gson();
    
    @Override
    public void initialize() {}

    @Override
    public Event intercept(Event record) {
        byte[] payload = record.getBody();
        String content = new String(payload, StandardCharsets.UTF_8);
        
        try {
            JsonObject parsed = gson.fromJson(content, JsonObject.class);
            if (parsed.has("logTime") && !parsed.get("logTime").isJsonNull()) {
                long bizTimestamp = parsed.get("logTime").getAsLong();
                record.getHeaders().put("timestamp", String.valueOf(bizTimestamp));
            }
        } catch (Exception ignored) {
            // 解析失败时保留原有Header，确保数据不丢失
        }
        return record;
    }

    @Override
    public List<Event> intercept(List<Event> batch) {
        List<Event> processed = new ArrayList<>(batch.size());
        for (Event item : batch) {
            processed.add(intercept(item));
        }
        return processed;
    }

    @Override
    public void close() {}

    public static class Factory implements Interceptor.Builder {
        
        @Override
        public Interceptor create() {
            return new LogTimeExtractor();
        }

        @Override
        public void configure(Context ctx) {}
    }
}

部署配置要点

将编译后的JAR包放置于Flume的lib目录，并在配置文件中绑定拦截器：

a1.sources.kafkaSrc.interceptors = timeFix
a1.sources.kafkaSrc.interceptors.timeFix.type = com.example.flume.interceptor.LogTimeExtractor$Factory

关键设计考量

• 容错处理：JSON解析异常时静默跳过，避免数据中断
• 批量优化：复用List容量预分配，减少GC压力
• 字段映射：根据实际日志结构调整logTime字段名