[Java知识库]2021.12.22flume拦截器演示InterceptorDemo

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;


/**
 * 对Source接收到event进行分辨
 * Event对象，：header，body
 * body内容 以hello开头，则给event头部header打入hello标签
 * body内容 以hi开头，则给event头header打入hi标签
 */

public class InterceptorDemo implements Interceptor {

    ArrayList<Event> addHeaderEvents=null;

    @Override
    public void initialize() {
        addHeaderEvents=new ArrayList<>();

    }

    @Override  //intercept拦截
    public Event intercept(Event event) {
        Map<String, String> headers = event.getHeaders();
        byte[] body = event.getBody();
        String bodyStr = new String(body);
        if (bodyStr.startsWith("hello"))
            headers.put("type","hello");
        else if (bodyStr.startsWith("hi"))
            headers.put("type","hi");
        else
            headers.put("type","other");
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> list) {
        addHeaderEvents.clear();
        for (Event event:list){
            Event opEvent = intercept(event);
            addHeaderEvents.add(opEvent);
        }

        return addHeaderEvents;
    }

    @Override
    public void close() {
        addHeaderEvents.clear();
        addHeaderEvents=null;

    }

    public static class Bulider implements Interceptor.Builder{
        @Override
        public Interceptor build() {
            return new InterceptorDemo();
        }

        @Override
        public void configure(Context context) {


        }
    }
}

[root@linux01 kb15conf]# vi ./netcat-flume-interceptor.conf 

interceptordemo.sources=interceptorDemoSource
interceptordemo.channels=helloChannel hiChannel otherChannel
interceptordemo.sinks=helloSink hiSink otherSink

interceptordemo.sources.interceptorDemoSource.type=netcat
interceptordemo.sources.interceptorDemoSource.bind=localhost
interceptordemo.sources.interceptorDemoSource.port=44444
interceptordemo.sources.interceptorDemoSource.interceptors=myinterceptors
interceptordemo.sources.interceptorDemoSource.interceptors.myinterceptors.type=nj.zb.kb15.InterceptorDemo$Bulider
interceptordemo.sources.interceptorDemoSource.selector.type=multiplexing
interceptordemo.sources.interceptorDemoSource.selector.mapping.hello=helloChannel
interceptordemo.sources.interceptorDemoSource.selector.mapping.hi=hiChannel
interceptordemo.sources.interceptorDemoSource.selector.mapping.other=otherChannel
interceptordemo.sources.interceptorDemoSource.selector.header=type

interceptordemo.channels.helloChannel.type=memory

interceptordemo.channels.hiChannel.type=memory

interceptordemo.channels.otherChannel.type=memory

interceptordemo.sinks.helloSink.type=hdfs
interceptordemo.sinks.helloSink.hdfs.fileType=DataStream
interceptordemo.sinks.helloSink.hdfs.filePrefix=hello
interceptordemo.sinks.helloSink.hdfs.fileSuffix=.txt
interceptordemo.sinks.helloSink.hdfs.path=hdfs://192.168.111.131:9000/kb15file/hello/%Y-%m-%d
interceptordemo.sinks.helloSink.hdfs.useLocalTimeStamp=true
interceptordemo.sinks.helloSink.hdfs.batchSize=640
interceptordemo.sinks.helloSink.hdfs.rollCount=0
interceptordemo.sinks.helloSink.hdfs.rollSize=6400000
interceptordemo.sinks.helloSink.hdfs.rollInterval=3

interceptordemo.sinks.hiSink.type=hdfs
interceptordemo.sinks.hiSink.hdfs.fileType=DataStream
interceptordemo.sinks.hiSink.hdfs.filePrefix=hi
interceptordemo.sinks.hiSink.hdfs.fileSuffix=.txt
interceptordemo.sinks.hiSink.hdfs.path=hdfs://192.168.111.131:9000/kb15file/hi/%Y-%m-%d
interceptordemo.sinks.hiSink.hdfs.useLocalTimeStamp=true
interceptordemo.sinks.hiSink.hdfs.batchSize=640
interceptordemo.sinks.hiSink.hdfs.rollCount=0
interceptordemo.sinks.hiSink.hdfs.rollSize=6400000
interceptordemo.sinks.hiSink.hdfs.rollInterval=3

interceptordemo.sinks.otherSink.type=logger

interceptordemo.sources.interceptorDemoSource.channels=helloChannel  hiChannel otherChannel
interceptordemo.sinks.helloSink.channel=helloChannel
interceptordemo.sinks.hiSink.channel=hiChannel
interceptordemo.sinks.otherSink.channel=otherChannel

[root@linux01 flume160]#./bin/flume-ng agent --name interceptordemo --conf ./conf/ --conf-file ./conf/kb15conf/netcat-flume-interceptor.conf -Dflume.root.logger=INFO,console

[root@linux01 conf]# telnet localhost 44444(后)