import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.charset.Charset;
import java.util.List;
/*
* 此程序为flume的拦截器,实现将flume采集到的数据转码功能
*/
public class MyTransCodingInterceptor implements Interceptor {
private static final Logger logger = LoggerFactory.getLogger(MyTransCodingInterceptor.class);
//当前编码
private static final Charset gbkCharset = Charset.forName("GB2312");
//需要转成的编码
private static final Charset gbkCharset1 = Charset.forName("UTF-8");
private static final String GBKCHAR = "GBK";
protected MyTransCodingInterceptor(Context context) {
logger.info("---------------CharsetTransferInterceptor---------------");
}
@Override
public void close() {
}
@Override
public void initialize() {
}
@Override
public Event intercept(Event event) {
byte[] body = event.getBody();
String strCharset = event.getHeaders().get("charset");
//转码时首先往回转一下,然后在继续转成需要的
String bodyStr = new String(body, gbkCharset);
String bodyStr1 = new String(bodyStr.getBytes(), gbkCharset1);
event.setBody(bodyStr1.getBytes());
return event;
}
@Override
public List<Event> intercept(List<Event> events) {
for (Event event : events) {
intercept(event);
}
return events;
}
public static class Builder implements Interceptor.Builder {
private Context context;
@Override
public void configure(Context context) {
this.context = context;
}
@Override
public Interceptor build() {
return new MyTransCodingInterceptor(context);
}
}
}
flume自定义拦截器实现采集数据转换编码
最新推荐文章于 2021-11-13 21:16:05 发布