对于日志,flume sink写入HDFS时,如果指定按照时间生成文件,在没有明确指定时间的情况下,会读取服务器时间作为创建文件的已经,这会导致日志的实际生成日期与文件不符。
这种情况下,可以通过拦截器在flume事件头指定timestamp作为文件的创建依据。
所谓零点漂移,就是上述问题的具体表现。即在按天生成日志文件的情况下,一条23:59:59左右生成的日志发送到服务器后可能已经是第二天了,如果没有指定时间,会被写入第二天对应的文件中,这就是所谓的零点漂移。
import com.alibaba.fastjson.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; import java.util.Map; public class TimeStampInterceptor implements Interceptor { private ArrayList<Event> events = new ArrayList<>(); @Override public void initialize() { } @Override public Event intercept(Event event) { Map<String, String> headers = event.getHeaders(); String log = new String(event.getBody(), StandardCharsets.UTF_8); JSONObject jsonObject = JSONObject.parseObject(log); String ts = jsonObject.getString("ts"); headers.put("timestamp", ts); return event; } @Override public List<Event> intercept(List<Event> list) { events.clear(); for (Event event : list) { events.add(intercept(event)); } return events; } @Override public void close() { } public static class Builder implements Interceptor.Builder { @Override public Interceptor build() { return new TimeStampInterceptor(); } @Override public void configure(Context context) { } } }