flume定制Interceptor
实现接口
接口:org.apache.flume.interceptor.Interceptor
方法1:Event intercept(Event var1);
Event是flume处理数据的数据结构,消息以byte存在Event的body中
该方法实现对每个消息处理逻辑,必须实现。
方法2:List<Event> intercept(List<Event> var1);
该方法实现对批量消息处理,flume在批量拉取消息时会通过该方法处理,必须实现
提供Interceptor实例访问
public static class Builder implements Interceptor.Builder { public Interceptor build() { return new SimpleDataTransformInterceptor(); } public void configure(Context context) { } }
Builder实现必须在我们的Interceptor类内部作为内部类
配置生效
在flume的properties配置文件中添加如下配置,使拦截器生效。
${agent_name}.sources.${source_name}.interceptors = ${interceptor_name}
${agent_name}.interceptors.${interceptor_name}.type=com.test.interceptor.MyInterceptor$Builder