Flume自定义SourceSink和Interceptor(简单功能实现)

Posted 2020-10-21 qwangxiao

　　1.Event
　　
　　event是flume传输的最小对象，从source获取数据后会先封装成event，然后将event发送到channel，sink从channel拿event消费。
　　
　　event由头headers和身体(body)两部分组成：Headers部分是一个map，body部分可以是String或者byte[]等。其中body部分是真正存放数据的地方，headers部分用于本节所讲的interceptor。
　　
　　2.Source
　　
　　自定义Source，自定义的Event需要继承PollableSource （轮训拉取）或者EventDrivenSource （事件驱动），另外还需要实现Configurable接口。
　　
　　PollableSource或者EventDrivenSource的区别在于：PollableSource是通过线程不断去调用process方法，主动拉取消息，而EventDrivenSource是需要触发一个调用机制，即被动等待。 Configurable接口：便于项目中初始化某些配置用的。
　　
　　Event:
　　
　　event是flume传输的最小对象，从source获取数据后会先封装成event，然后将event发送到channel，sink从channel拿event消费。
　　
　　2.1CustomSource.java
　　
　　public class CustomSource extends AbstractSource implements Configurable,PollableSource{
　　
　　@Override
　　
　　public long getBackOffSleepIncrement() {
　　
　　// TODO Auto-generated method stub
　　
　　return 0;
　　
　　}
　　
　　@Override
　　
　　public long getMaxBackOffSleepInterval() {
　　
　　// TODO Auto-generated method stub
　　
　　return 0;
　　
　　}
　　
　　@Override
　　
　　public Status process() throws EventDeliveryException {
　　
　　Random random = new Random();
　　
　　int randomNum = random.nextInt(100);
　　
　　String text = "Hello world" + random.nextInt(100);
　　
　　HashMap<String, String> header = new HashMap<String,String>();
　　
　　header.put("id",Integer.toString(randomNum));
　　
　　this.getChannelProcessor()
　　
　　.processEvent(EventBuilder.withBody(text,Charset.forName("UTF-8"),header));
　　
　　return Status.READY;
　　
　　}
　　
　　@Override
　　
　　public void configure(Context arg0) {
　　
　　2.2Flume 编写配置文件：
　　
　　# 指定Agent的组件名称
　　
　　a1.sources = r1
　　
　　a1.sinks = k1
　　
　　a1.channels = c1
　　
　　# 指定Flume source(要监听的路径)
　　
　　a1.sources.r1.type = com.harderxin.flume.test.MySource
　　
　　# 指定Flume sink
　　
　　a1.sinks.k1.type = file_roll
　　
　　# sink的输出目录，根据自己情况定义
　　
　　a1.sinks.k1.sink.directory = www.jyyL157.com /home/hadoop/sinkFolder
　　
　　# 指定Flume channel
　　
　　a1.channels.c1.type = memory
　　
　　a1.channels.c1.capacity = 1000
　　
　　a1.channels.c1.transactionCapacity = 100
　　
　　a1.channels.c1.byteCapacityBufferPercentage = 20
　　
　　a1.channels.c1.byteCapacity = 800000
　　
　　# 绑定source和sink到channel上
　　
　　a1.sources.r1.channels = c1
　　
　　2.3打jar包运行：
　　
　　将项目打成jar包（可以只打程所在的类），放入flume下的 lib目录下（网上说是bin目录，但没有运行成功）。
　　
　　然后bin目录下执行：
　　
　　flume-ng agent -conf conf -conf-file ..www.dfgjpt.com /conf/custom_source.conf -name a1
　　
　　1
　　
　　3.Sink
　　
　　3.1CustomSink.java
　　
　　public class CustomSink extends AbstractSink implements Configurable{
　　
　　@Override
　　
　　public Status process() throws EventDeliveryException {
　　
　　Status status = Status.READY;
　　
　　Transaction trans = null;
　　
　　try {
　　
　　Channel channel = getChannel();
　　
　　trans = channel.getTransaction();
　　
　　trans.begin();
　　
　　for(int i= 0;i < 100 ;i++) {
　　
　　Event event = channel.take();
　　
　　if(event == null) {
　　
　　status = status.BACKOFF;
　　
　　break;
　　
　　}else {
　　
　　String body = new String(event.getBody());
　　
　　System.out.println(body);
　　
　　}
　　
　　}
　　
　　trans.commit();
　　
　　}catch (Exception e) {
　　
　　if(trans != null) {
　　
　　trans.commit();
　　
　　}
　　
　　e.printStackTrace(www.120xh.cn/);
　　
　　}finally {
　　
　　if(trans != null) {
　　
　　trans.close(chuangshi88.cn);
　　
　　}
　　
　　}
　　
　　return status;
　　
　　}
　　
　　@Override
　　
　　public void configure(Context arg0) {
　　
　　3.2custom_sink.conf
　　
　　a1.sources = r1
　　
　　a1.sinks = k1
　　
　　a1.channels = c1
　　
　　# Describe/configure the source
　　
　　a1.sources.r1.type = netcat
　　
　　a1.sources.r1.bind = 0.0.0.0
　　
　　a1.sources.r1.port = 44444
　　
　　# Describe the sink
　　
　　a1.sinks.k1.type = com.caoxufeng.MyCustom.CustomSink
　　
　　# Use a channel which buffers events in memory
　　
　　a1.channels.c1.type = memory
　　
　　a1.channels.c1.capacity = 10000
　　
　　a1.channels.c1.transactionCapacity = 100
　　
　　# Bind the source and sink to the channel
　　
　　a1.sources.r1.channels = c1
　　
　　a1.sinks.k1.channel = c1
　　
　　3.3打jar包运行：
　　
　　将项目打成jar包（可以只打程所在的类），放入flume下的 lib目录下（网上说是bin目录，但没有运行成功）。
　　
　　然后bin目录下执行：
　　
　　flume-ng agent -conf conf -conf-file ../conf/custom_sink.conf -name a1
　　
　　1
　　
　　4.Interceptor
　　
　　用户Source读取events发送到Sink的时候，在events header中加入一些有用的信息，或者对events的内容进行过滤，完成初步的数据清洗。
　　
　　Fluem实现了日志的多来源自动抽取和多target的自动发送等功能。一直以来人们都是将数据清洗的过程放在Hadoop 的 MR的进行的。而自定义Interceptor可以让Flume进行数据清洗匹配，过滤到那些不规则的脏数据。
　　
　　Flume中拦截器的作用就是对于event中header的部分可以按需塞入一些属性，当然你如果想要处理event的body内容，也是可以的，但是event的body内容是系统下游阶段真正处理的内容，如果让Flume来修饰body的内容的话，那就是强耦合了，这就违背了当初使用Flume来解耦的初衷了。
　　
　　4.1CustomInterceptor.java
　　
　　public class CustomInterceptor implements Interceptor{
　　
　　private final String headerKey;
　　
　　private static final String CONF_HEADER_KEY = "header";
　　
　　private static final String DEFAULT_HEADER = "count";
　　
　　private final AtomicLong currentCount;
　　
　　public CustomInterceptor(Context www.wangcai157.com ctx) {
　　
　　headerKey = ctx.getString(CONF_HEADER_KEY,DEFAULT_HEADER);
　　
　　currentCount = new AtomicLong();
　　
　　}
　　
　　//运行前的初始化，一般不需要实现
　　
　　@Override
　　
　　public void initialize() {
　　
　　// TODO Auto-generated method stub
　　
　　}
　　
　　//)处理单个event
　　
　　@Override
　　
　　public Event intercept(Event event) {
　　
　　long count = currentCount.incrementAndGet();
　　
　　event.getHeaders(www.ylzx1980.com).put(headerKey, String.valueOf(count));
　　
　　return event;
　　
　　}
　　
　　//批量处理event，循环出路一面的interceptor(Event event)
　　
　　@Override
　　
　　public List<Event> intercept(List<Event> events) {
　　
　　for(Event e:events) {
　　
　　intercept(e);
　　
　　}
　　
　　return events;
　　
　　}
　　
　　@Override
　　
　　public void close() {
　　
　　}
　　
　　public static class CounterInterceptorBuilder implements Builder {
　　
　　private Context ctx;
　　
　　@Override
　　
　　public Interceptor build() {
　　
　　return new CustomInterceptor(ctx);
　　
　　}
　　
　　@Override
　　
　　public void configure(Context context) {
　　
　　this.ctx = context;
　　
　　方法intercept(Event www.caibaoyule.cn event)是具体执行解析的方法，将count自增1，然后写入到该条event的headers中。
　　
　　4.2custom_interceptor.conf
　　
　　a1.sources = r1
　　
　　a1.sinks = s1
　　
　　a1.channels = c1
　　
　　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.caoxufeng.MyCustom.CustomInterceptor$CounterInterceptorBuilder
　　
　　a1.sources.r1.interceptors.i1.perserveExisting = true
　　
　　a1.sinks.s1.type = logger
　　
　　a1.channels.c1.type = memory
　　
　　a1.channels.c1.capacity = 2
　　
　　a1.channels.c1.transactionCapacity = 2
　　
　　a1.sources.r1.channels = c1
　　
　　a1.sinks.s1.channel = c1
　　
　　4.3打jar包运行：
　　
　　将项目打成jar包（可以只打程所在的类），放入flume下的 lib目录下（网上说是bin目录，但没有运行成功）。
　　
　　然后bin目录下执行：
　　
　　flume-ng agent -c conf -f ../conf/custom-interceptor.conf -n a1