采集数据零点漂移问题解析

为什么会发生零点漂移？

比如2021-10-10 23:59:59生成的日志文件，然后数据经过第一层的flume采集，加上kafka的缓冲，然后到 集群的另一台上的第二层的flume的时候，时间肯定就会到2020-10-11 00:00:XX了，这样一来，如果采用当前系统时间作为timestamp的话，2020-10-10 的日志数据就会上传到hdfs上的2020-10-10 的目录下。因为Kafka Source会为其加上该header，value为当前系统的时间戳Kafka Source会为其加上该header，value为当前系统的时间戳

所以我们需要再第二层flume里面写这么一个加时间戳的拦截器，把日志文件里面的时间添加到event的header里面

public class TimestampInterceptor implements Interceptor {
    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {

        //获取Event中的body,并转为String字符串
        byte[] body = event.getBody();
        String log = new String(body, StandardCharsets.UTF_8);

        //解析日志，得到json对象
        JSonObject jsonObject = JSON.parseObject(log);

        if(jsonObject.containsKey("ts")){
            //获取json对象中的ts字段，即为日志的时间戳
            String ts = jsonObject.getString("ts");

            //为event加入timestamp的header
            event.getHeaders().put("timestamp",ts);

        }

        return event;
    }

    @Override
    public List intercept(List list) {

        for (Event event : list) {
            intercept(event);
        }
        return list;
    }

    @Override
    public void close() {

    }

    public static class Builder implements Interceptor.Builder{

        @Override
        public Interceptor build() {
            return new TimestampInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }
}