flume开发-自定义拦截器(Interceptor)

谁践踏了优雅 2022-06-14 01:43 302阅读 0赞

拦截器是简单的插件式组件，设置在source和channel之间。source接收到的时间，在写入channel之前，拦截器都可以进行转换或者删除这些事件。每个拦截器只处理同一个source接收到的事件。flume官方实现了很多拦截器也可以自定义拦截器。通过实现自定义的拦截器可以对日志进行ETL。

自定义拦截器只需要实现Interceptor的继承类。具体步骤如下：

1. 实现一个Interceptor的继承类。下面的example实现了继承类ETLInterceptor. package 是com.test.flume .

**\[java\]**  [view plain][]   [copy][view plain]

1.  //add begin  
2.  **package** com.test.flume;  
3.  //add end  
4.  **import** com.google.common.base.Charsets;  
5.  **import** com.google.common.collect.Lists;  
6.    
7.  **import** org.apache.flume.Context;  
8.  **import** org.apache.flume.Event;  
9.  **import** org.apache.flume.interceptor.Interceptor;  
10. **import** org.slf4j.Logger;  
11. **import** org.slf4j.LoggerFactory;  
12.   
13. **import** java.util.Arrays;  
14. **import** java.util.List;  
15.   
16.   
17. clog.sources.source\_log3.interceptors.i3.type=com.thextrader.dmp.streaming.flume.BidInfoLogUrlFilter$Builder  
18.   
19.   
20. **public** **class** ETLInterceptor **implements** Interceptor\{  
21.     **private** **static** Logger logger = LoggerFactory.getLogger(AuctionETLInterceptor.**class**);  
22.     @Override  
23.     **public** **void** initialize() \{  
24.   
25.     \}  
26.     @Override  
27.     **public** Event intercept(Event event) \{  
28.         String body = **new** String(event.getBody(), Charsets.UTF\_8);  
29.         String newBody=body;  
30.         **try**\{  
31.        //add begin  
32.             String\[\] splits = body.split("\\\\^");  
33.             splits\[6\]="";  
34.             splits\[9\]="";  
35.             splits\[10\]="";  
36.             splits\[11\]="";  
37.             splits\[17\]="";  
38.             splits\[25\]="";  
39.             **if**(splits.length>28)\{  
40.                 newBody = String.join("^", Arrays.copyOfRange(splits,0,28));  
41.             \}**else**\{  
42.                 newBody = String.join("^", splits);  
43.             \}  
44.        //add end  
45.             event.setBody(newBody.toString().getBytes());  
46.         \}**catch** (Exception e)\{  
47.             logger.warn(body,e);  
48.             event=**null**;  
49.         \}  
50.         **return** event;  
51.     \}  
52.   
53.     @Override  
54.     **public** List<Event> intercept(List<Event> events) \{  
55.         List<Event> intercepted = Lists.newArrayListWithCapacity(events.size());  
56.         **for** (Event event : events) \{  
57.             Event interceptedEvent = intercept(event);  
58.             **if** (interceptedEvent != **null**) \{  
59.                 intercepted.add(interceptedEvent);  
60.             \}  
61.         \}  
62.         **return** intercepted;  
63.     \}  
64.   
65.     @Override  
66.     **public** **void** close() \{  
67.   
68.     \}  
69.   
70.     **public** **static** **class** Builder **implements** Interceptor.Builder\{  
71.         //使用Builder初始化Interceptor  
72.         @Override  
73.         **public** Interceptor build() \{  
74.     //add begin  
75.             **return** **new** ETLInterceptor();  
76.     //add end  
77.         \}  
78.   
79.         @Override  
80.         **public** **void** configure(Context context) \{  
81.   
82.         \}  
83.     \}  
84. \}

public Event intercept(Event event) 函数中写你需要的ETL等逻辑。

public static class Builder implements Interceptor.Builder 函数中new 出继承类 ETLInterceptor。

2. 将上面的java代码打成jar包。

在flume的安装目录下的plugins.d 目录下新建文件夹ETLInterceptor.文件夹这种新建三个文件夹lib，libext，native。

将jar包放入lib文件夹中。

3. 配置flume source的interceptor type为com.test.flume.ETLInterceptor.$Builder

4.启动flume ，自定义的拦截器就生效了。

[view plain]: http://blog.csdn.net/yanshu2012/article/details/54019034#