Flume正则拦截器(regex_filter) 以及自定义拦截器

╰半橙微兮° 2022-11-26 06:59 240阅读 0赞

前面已经给大家说过flume的简介以及安装，还有不明白的可以再去看看：  
[Flume简介以及详细安装教程][Flume]

在当今大数据时代，在处理海量数据之前，收集数据，聚合和转换数据是绝对必要的，并最终将数据移动到那些使用不同分析和数据挖掘工具的存储库中。  
执行所有这些步骤的流行工具之一是Apache Flume。 这些数据通常是以事件或日志的形式存储。 Apache Flume有三个主要组件：

**Source**：数据源可以是企业服务器，文件系统，云端，数据存储库等。  
**Channel**：在事件被sink消耗前由Channel 存储。 Channel 是被动存储。 Channel 支持故障恢复和高可靠性; Channel 示例是由本地文件系统和基于内存的Channel 支持的文件通道。  
**Sink**：Sink是可以存储数据的目标存储库。 它可以是一个集中的地方，如HDFS，像Apache Spark这样的处理引擎，或像ElasticSearch这样的数据存储库/搜索引擎。

Flume是高度可配置的，并且支持许多源，channel，serializer和sink。它还支持数据流。 Flume的强大功能是拦截器，支持在运行中修改/删除事件的功能。支持的拦截器之一是regex\_filter。

**regex\_filter**将事件体解释为文本，并将其与提供的正则表达式进行对比，并基于匹配的模式和表达式，包括或排除事件。我们将详细看看regex\_filter。

**案例**：监听文件ss.txt，过滤掉以userid为开头的字段。  
要监听的文件放在opt/sofft/datas目录下  
ss.txt的文件内容如下：

userid username
    hello hadoop
    hello hive
    hadoop hive
    hello world
    userid hello
    hello cm
    hello hadoop
    hello hive

编写flume配置文件：  
vi /opt/flumeconf/conf\_0810\_interceptor.properties

a3.channels=c3
    a3.sources=s3
    a3.sinks=k3
    
    #定义具体的source类型
    a3.sources.s3.type=spooldir
    #定义监听的文件目录
    a3.sources.s3.spoolDir=/opt/soft/datas
    a3.sources.s3.interceptors=userid_filter
    
    #定义正则拦截器filter
    a3.sources.s3.interceptors.userid_filter.type=regex_filter
    #定义正则，要拦截的字段
    a3.sources.s3.interceptors.userid_filter.regex=userid.*
    #满足正则就过滤掉当定义是true时，默认是false
    a3.sources.s3.interceptors.userid_filter.excludeEvents=true
    
    a3.sources.s3.deletePolicy=immediate
    a3.sources.s3.pollDelay=500
    
    a3.channels.c3.type=memory
    a3.sinks.k3.type= logger
    
    a3.sources.s3.channels=c3
    a3.sinks.k3.channel = c3

**注意：**  
pooldir默认是将处理完的新文件重命名，原文件名会自动添加后缀名.COMPLETE，但是遇到同名文件之后，处理完同名文件，在重命名时，发现已经有相同文件名的文件存在，此时flume会报错，然后停止监控线程（此线程默认500毫秒扫描一次目录，检查是否有新文件），目录中再有新的文件，也不会被处理。  
处理方法为：  
设置spooldir文件删除策略为：**a3.sources.s1.deletePolicy=immediate**

**启动flume**

flume-ng agent -n a3 -c conf -f /opt/flumeconf/conf_0810_interceptor.properties -Dflume.root.logger=INFO,console

这时被监听的文件内如果有userid为开头的字段就会被过滤掉！  
如下图所示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70][]  
当文件读取完毕时，原文件名会自动添加后缀名.COMPLETE  
如下图所示：  
![202008110158050.png][]

### 自定义拦截器 ###

**单个事件拦截**  
案例：  
文件中出现“男”，就用1代替，文件中出现“女”，就用2代替，剩余的就用0代替  
文件原始内容：  
001，张三，男，24  
002，李四，女，18  
003，王五，women,34  
… 类似这样的格式  
处理之后的内容：  
001，张三，1，24  
002，李四，1，18  
003，王五，0，34

自定义拦截器代码：

先导入依赖

<dependency>
        <groupId>org.apache.flume</groupId>
        <artifactId>flume-ng-core</artifactId>
        <version>1.6.0</version>
    </dependency>

自定义拦截器代码：  
创建CustomInterceptor类implements Interceptor  
重写方法：

import org.apache.flume.Context;
    import org.apache.flume.Event;
    import org.apache.flume.interceptor.Interceptor;
    
    import java.util.List;
    
    public class CustomInterceptor implements Interceptor { 
        @Override
        public void initialize() { 
    
       }
       //单个事件拦截
      @Override
      public Event intercept(Event event) { 
        byte[] body = event.getBody();
        String line = new String(body);  //001,张三，男,24
        String[] sps = line.split(",");
        switch (sps[2]){ 
        	case "男" :
        		sps[2] = "1";
        		break;
        	case "女":
       		sps[2] = "2";
        		break;
        	default:
                	sps[2] = "0";
                	break;
    	}
        String newstr = sps[0]+","+sps[1]+","+sps[2]+","+sps[3];
        event.setBody(newstr.getBytes());
        //自定义头信息
        return event;
      }
      //批量事件拦截
      @Override
      public List<Event> intercept(List<Event> list) { 
          for (Event event : list) { 
                intercept(event);
          }
           return list;
       }
      @Override
      public void close() { 
    
      }
      
      //创建类Builder实现Interceptor.Builder
      public static class Builder implements Interceptor.Builder{ 
    
        @Override
        public Interceptor build() { 
            return null;
        }
         @Override
         public void configure(Context context) { 
           
            }
        }
    }

编写flume配置文件：  
vi /opt/flumeconf/conf\_0810\_custconf.properties

a4.channels=c4
    a4.sources=s4
    a4.sinks=k4
    
    a4.sources.s4.type=spooldir
    a4.sources.s4.spoolDir=/opt/soft/datas
    a4.sources.s4.interceptors=myintec
    #绑定自定义拦截器的类名
    a4.sources.s4.interceptors.myintec.type=com.demo.interceports.filter.CustomInterceptor$Builder
    
    a4.channels.c4.type=memory
    a4.sinks.k4.type=logger
    
    a4.sinks.k4.channel=c4
    a4.sources.s4.channels=c4

启动flume

flume-ng agent -n a4 -c conf -f /opt/flumeconf/conf_0810_custconf.properties -Dflume.root.logger=INFO,console

[Flume]: https://blog.csdn.net/zp17834994071/article/details/107895044
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70]: /images/20221124/9f7020678e5a4dbfa3168efc332845dc.png
[202008110158050.png]: /images/20221124/dec24ca0e92c4b138ace1711b9e6ff88.png