Flume 自定义Interceptor

Flume 自定义Interceptor Flume 自定义Interceptor

1）案例需求
使用 Flume 采集服务器本地日志，需要按照日志类型的不同，将不同种类的日志发往不同的分析系统。
2）需求分析

在实际的开发中，一台服务器产生的日志类型可能有很多种，不同类型的日志可能需要发送到不同的分析系统。此时会用到 Flume 拓扑结构中的 Multiplexing 结构，Multiplexing的原理是，根据 event 中 Header 的某个 key 的值，将不同的 event 发送到不同的 Channel中，所以我们需要自定义一个 Interceptor，为不同类型的 event 的 Header 中的 key 赋予不同的值。

在该案例中，我们以端口数据模拟日志，以是否包含”atguigu”模拟不同类型的日志，我们需要自定义 interceptor 区分数据中是否包含”atguigu”，将其分别发往不同的分析系统（Channel）。

Interceptor和Multiplexing ChannelSelector案例

3）实现步骤
（1）创建一个 maven 项目，并引入以下依赖。

 org.apache.flume
 flume-ng-core
 1.9.0

（2）定义 CustomInterceptor 类并实现 Interceptor 接口。

public class TypeInterceptor implements Interceptor {
 	//声明一个存放事件的集合
 	private List addHeaderEvents;
	@Override
	public void initialize() {
 		//初始化存放事件的集合
 		addHeaderEvents = new ArrayList<>();
 	}
 	
 	//单个事件拦截
 	@Override
	public Event intercept(Event event) {
 		//1.获取事件中的头信息
 		Map headers = event.getHeaders();
 		//2.获取事件中的 body 信息
 		String body = new String(event.getBody());
 		//3.根据 body 中是否有"atguigu"来决定添加怎样的头信息
 		if (body.contains("atguigu")) {
 			//4.添加头信息
 			headers.put("type", "first");
 		} else {
 			//4.添加头信息
 			headers.put("type", "second");
 		}
 		return event;
 	}
	
	//批量事件拦截
	 @Override
 	public List intercept(List events) {
 		//1.清空集合
 		addHeaderEvents.clear();
 		//2.遍历 events
 		for (Event event : events) {
 			//3.给每一个事件添加头信息
 			addHeaderEvents.add(intercept(event));
 		}
 		//4.返回结果
 		return addHeaderEvents;
 	}
 	@Override
 	public void close() {
 	}
 	public static class Builder implements Interceptor.Builder {
	 	@Override
 		public Interceptor build() {
			 return new TypeInterceptor();
		 }
 		@Override
		public void configure(Context context) {
		}
	}
}

（3）编辑 flume 配置文件
为 hadoop102 上的 flume1 配置 1 个 netcat source，1 个 sink group（2 个 avro sink），并配置相应的 ChannelSelector 和 interceptor。

# Name the components on this agent
a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1 c2
# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.xiaoqiu.TypeInterceptor$Builder
a1.sources.r1.selector.type = multiplexing
a1.sources.r1.selector.header = type
a1.sources.r1.selector.mapping.first = c1
a1.sources.r1.selector.mapping.second = c2
# Describe the sink
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = hadoop103
a1.sinks.k1.port = 4141
a1.sinks.k2.type=avro
a1.sinks.k2.hostname = hadoop104
a1.sinks.k2.port = 4242
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Use a channel which buffers events in memory
a1.channels.c2.type = memory
a1.channels.c2.capacity = 1000
a1.channels.c2.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1 c2
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c2

为 hadoop103 上的 flume2 配置一个 avro source 和一个 logger sink。

a1.sources = r1
a1.sinks = k1
a1.channels = c1
a1.sources.r1.type = avro
a1.sources.r1.bind = hadoop103
a1.sources.r1.port = 4141
a1.sinks.k1.type = logger
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
a1.sinks.k1.channel = c1
a1.sources.r1.channels = c1

为 hadoop104 上的 flume3 配置一个 avro source 和一个 logger sink。

a1.sources = r1
a1.sinks = k1
a1.channels = c1
a1.sources.r1.type = avro
a1.sources.r1.bind = hadoop104
a1.sources.r1.port = 4242
a1.sinks.k1.type = logger
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
a1.sinks.k1.channel = c1
a1.sources.r1.channels = c1

（4）分别在 hadoop102，hadoop103，hadoop104 上启动 flume 进程，注意先后顺序。
[hadoop@hadoop102 flume]$ bin/flume-ng agent --conf conf/ --name a1 --conf-file job/flume1.conf
[hadoop@hadoop103 flume]$ bin/flume-ng agent --conf conf/ --name a1 --conf-file job/flume2.conf -Dflume.root.logger=INFO,console
[hadoop@hadoop104 flume]$ bin/flume-ng agent --conf conf/ --name a1 --conf-file job/flume3.conf -Dflume.root.logger=INFO,console
（5）在 hadoop102 使用 netcat 向 localhost:44444 发送字母和数字。
（6）观察 hadoop103 和 hadoop104 打印的日志。