返回override复习flumel对接kafka.pdfVIP

1.Flume器代码

packagecom.atguigu.flume.topic;

importorg.apache.flume.Context;

importorg.apache.flume.Event;

importjava.util.List;

/**

*@authorliubo

*/

publicclassInterceptorimplements

org.apache.flume.interceptor.Interceptor{

@Override

publicvoidinitialize(){

}

@Override

publicEventintercept(Eventevent){

if(1=event.getBody()[0]event.getBody()[0]=9){

event.getHeaders().put(topic,number);

}elseif(a=event.getBody()[0]event.getBody()[0]=z){

event.getHeaders().put(topic,letter);

}

returnevent;

}

@Override

publicListEventintercept(ListEventevents){

returnnull;

}

@Override

publicvoidclose(){

}

publicstaticclassCustomBuilderimplements

Interceptor.Builder{

@Override

publicorg.apache.flume.interceptor.Interceptorbuild(){

returnnewInterceptor();

}

@Override

publicvoidconfigure(Contextcontext){

}

}

}

2.Flume配置文件

#define

a1.sources=r1

a1.sinks=k1

a1.channels=c1

#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.atguigu.flume.topic.Interceptor$CustomBuilder

#sink

a1.sinks.k1.type=org.apache.flume.sink.kafka.KafkaSink

a1.sinks.k1.kafka.bootstrap.servers=

hadoop102:9092,hadoop103:9092,hadoop104:9092

a1.sinks.k1.kafka.flumeBatchSize=20

a1.sinks.k1.kafka.producer.acks=1

a1.sinks.k1.kafka.producer.linger.