五月婷婷丁香性爱|j久久一级免费片|久久美女福利视频|中文观看在线观看|加勒比四区三区二|亚洲裸女视频网站|超碰97AV在线69网站免费观看|有码在线免费视频|久久青青日本视频|亚洲国产AAAA

自定義攔截器處理日志

在Flume中自定義攔截器的方式主要流程如下（以java語言為例），以下示例中的FLUME_HOME表示Flume的安裝路徑，例如/tools/flume（僅供參考），實(shí)際配置的時候，請以用戶安裝Flume的實(shí)際路徑為準(zhǔn)。

1. 創(chuàng)建MAVEN工程項(xiàng)目，引入Flume依賴。

    

   根據(jù)集群中的 Flume 版本，引入 Flume 依賴，如下所示：

   <dependencies>
           <dependency>
               <groupId>org.apache.flume</groupId>
               <artifactId>flume-ng-core</artifactId>
               <version>1.10.1</version>
               <scope>provided</scope>
           </dependency>
       </dependencies>

   無需將該依賴打包進(jìn)最后的JAR包中，故將其作用域設(shè)置為provided。

    

2. 創(chuàng)建類實(shí)現(xiàn)攔截器接口Interceptor，并且實(shí)現(xiàn)相關(guān)方法。

    

   • initialize() 方法：初始化攔截器操作，讀取配置信息、建立連接等。
   • intercept(Event event) 方法：用于攔截單個事件，并對事件進(jìn)行處理。接收一個事件對象作為輸入，并返回一個修改后的事件對象。
   • intercept(List<Event> list) 方法：事件批處理，攔截事件列表，并對事件列表進(jìn)行處理。
   • close() 方法：關(guān)閉攔截器，在這里釋放資源、關(guān)閉連接等。

   import org.apache.flume.Event;
   import org.apache.flume.interceptor.Interceptor;
   
   import java.nio.charset.StandardCharsets;
   import java.util.ArrayList;
   import java.util.List;
   
   public class TestInterceptor implements Interceptor {
       @Override
       public void initialize() {
   
       }
       @Override
       public Event intercept(Event event) {
   
           // 獲取事件數(shù)據(jù)
           String eventData = new String(event.getBody(), StandardCharsets.UTF_8);
           // 檢查事件數(shù)據(jù)中是否包含指定字符串
           if (eventData.contains("hello")) {
               // 如果包含指定字符串，則過濾掉該事件，返回 null
               return null;
           }
   
           return event;
       }
   
       @Override
       public List<Event> intercept(List<Event> events) {
           // 創(chuàng)建一個新的列表，存儲處理過后的事件
           List<Event> interceptedEvents = new ArrayList<>();
           for (Event event : events) {
               Event interceptedEvent = intercept(event);
               if (interceptedEvent != null) {
                   interceptedEvents.add(interceptedEvent);
               }
           }
           return interceptedEvents;
       }
   
       @Override
       public void close() {
   
       }
   
   }

3. 構(gòu)建攔截器，攔截器的創(chuàng)建和配置通常是通過 Builder 模式來完成的，完整的代碼如下所示：

    

   import org.apache.flume.Context;
   import org.apache.flume.Event;
   import org.apache.flume.interceptor.Interceptor;
   
   import java.nio.charset.StandardCharsets;
   import java.util.ArrayList;
   import java.util.List;
   
   public class TestInterceptor implements Interceptor {
       @Override
       public void initialize() {
       }
       @Override
       public Event intercept(Event event) {
           // 獲取事件數(shù)據(jù)
           String eventData = new String(event.getBody(), StandardCharsets.UTF_8);
           // 檢查事件數(shù)據(jù)中是否包含指定字符串
           if (eventData.contains("hello")) {
               // 如果包含指定字符串，則過濾掉該事件，返回 null
               return null;
           }
           return event;
       }
       @Override
       public List<Event> intercept(List<Event> events) {
           List<Event> interceptedEvents = new ArrayList<>();
           for (Event event : events) {
               Event interceptedEvent = intercept(event);
               if (interceptedEvent != null) {
                   interceptedEvents.add(interceptedEvent);
               }
           }
           return interceptedEvents;
       }
   
       @Override
       public void close() {
   
       }
   
       // 攔截器構(gòu)建
       public static class Builder implements Interceptor.Builder {
   
           @Override
           public void configure(Context context) {
   
           }
   
           @Override
           public Interceptor build() {
               return new TestInterceptor();
           }
   
       }
   
   }

4. 轉(zhuǎn)換為jar包，并且將其上傳至Flume安裝路徑下的lib文件夾下（請以用戶安裝Flume的實(shí)際路徑為準(zhǔn)）。
5. 編寫配置文件，需要將自定義的攔截器配置進(jìn)去。

    

   攔截器全類名配置時需要注意，格式為攔截器的全類名 + $Builder。

   # 攔截器配置
   # 攔截器定義
   a1.sources.r1.interceptors = i1
   # 攔截器全類名
   a1.sources.r1.interceptors.i1.type = TestInterceptor$Builder

6. 運(yùn)行Flume即可。

public class TestInterceptor implements Interceptor {
    @Override
    public void initialize() {
    }

    @Override
    public Event intercept(Event event) {
        // 獲取事件數(shù)據(jù)
        String eventData = new String(event.getBody(), StandardCharsets.UTF_8);
        Map<String, Object> splitMap = new HashMap<>();
        String[] splitList = eventData.split(" ");
        for (int i = 0; i < splitList.length; i++) {
            splitMap.put("field" + i, splitList[i].trim());
        }
        eventData.setBody(splitMap.toString().getBytes(StandardCharsets.UTF_8));
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        List<Event> interceptedEvents = new ArrayList<>();
        for (Event event : events) {
            Event interceptedEvent = intercept(event);
            if (interceptedEvent != null) {
                interceptedEvents.add(interceptedEvent);
            }
        }
        return interceptedEvents;
    }

    @Override
    public void close() {
    }
}