Hive之自定义拦截器

傷城~ 2022-12-26 02:29 3阅读 0赞

# Hive之自定义拦截器 #

## 一、概述 ##

**1）案例需求**

> 使用Flume采集服务器本地日志，需要按照日志类型的不同，将不同种类的日志发往不同的分析系统。

**2）需求分析**

> 在实际的开发中，一台服务器产生的日志类型可能有很多种，不同类型的日志可能需要发送到不同的分析系统。此时会用到Flume拓扑结构中的Multiplexing结构，Multiplexing的原理是，根据event中Header的某个key的值，将不同的event发送到不同的Channel中，所以我们需要自定义一个Interceptor，为不同类型的event的Header中的value赋予不同的值。  
> 在该案例中，我们以端口数据模拟日志，以sys和user开头的数据模拟不同类型的日志，我们需要自定义interceptor区分不同开头的日志数据，将其分别发往不同的分析系统（Channel），在HDFS上存储为不同文件夹下面。

## 二、代码开发 ##

**（1）创建一个maven项目，并引入以下依赖。**

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

**（2）定义CustomInterceptor类并实现Interceptor接口。**

package com.hadoop.flume;
    
    import org.apache.flume.Context;
    import org.apache.flume.Event;
    import org.apache.flume.interceptor.Interceptor;
    
    import java.util.List;
    import java.util.Map;
    
    public class MyInterceptor implements Interceptor { 
    
        //初始化
        public void initialize() { 
    
        }
    
        //操作方法
    
        /** * 针对单条数据进行处理（source没有batchsize的时候调用） * @param event * @return */
        public Event intercept(Event event) { 
    
            //1.取出数据
            String s = new String(event.getBody());
            //2.对数据进行切分并判断数据类型
            String logType = s.split(":")[0];
            //3.获取headers的索引信息
            Map<String, String> headers = event.getHeaders();
            //4.向headers中添加数据类型信息
            headers.put("logType",logType);
            //5.返回单个事务
            return event;
        }
    
        /** * 处理一个批次的数据（source有batchesize时调用） * 内部调用单个事件的处理方法 * @param list * @return */
        public List<Event> intercept(List<Event> list) { 
    
            for (Event event : list) { 
                intercept(event);
            }
    
            return list;
        }
    
        /** * 关闭资源 */
        public void close() { 
    
        }
    
    
        public static class Builder implements org.apache.flume.interceptor.Interceptor.Builder{ 
    
            /** * 提供flume构建自定义拦截器的对象，flume调用此方法对自定义拦截器进行构建 * @return */
            public Interceptor build() { 
                return new MyInterceptor();
            }
    
            /** * 获取配置文件的配置 * @param context */
            public void configure(Context context) { 
    
            }
        }
    
    }

（3）打包上传到Flume家目录下的lib文件夹中

## 三、启动Flume并测试 ##

1.  编写Flume脚本

#1、定义agent、source、channel、sink的名称
    a1.sources = r1
    a1.channels = c1 c2
    a1.sinks = k1 k2
    
    #2、描述source
    a1.sources.r1.type = netcat
    a1.sources.r1.bind = 0.0.0.0
    a1.sources.r1.port = 9999
    
    #2.1、描述channel selector
    a1.sources.r1.selector.type = multiplexing
    #设置channel selector使用header里面那个key
    a1.sources.r1.selector.header = logType
    #设置header对应的每个value值应该发到哪个channel
    a1.sources.r1.selector.mapping.sys = c1
    a1.sources.r1.selector.mapping.user = c2
    
    #2.2、描述拦截器
    #指定拦截器的名称
    a1.sources.r1.interceptors = i1
    #指定拦截器的全类名
    a1.sources.r1.interceptors.i1.type = com.atguigu.flume.MyInterceptor$Builder
    
    #3、描述channel
    #3.1、描述channel1
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 1000
    
    #3.2、描述channel2
    a1.channels.c2.type = memory
    a1.channels.c2.capacity = 1000
    a1.channels.c2.transactionCapacity = 1000
    
    #4、描述sink
    #4.1、描述sink1
    a1.sinks.k1.type = hdfs
    a1.sinks.k1.hdfs.path = hdfs://hadoop102:8020/flume/%{ logType}
    a1.sinks.k1.hdfs.filePrefix = sys_
    a1.sinks.k1.hdfs.rollInterval = 30
    a1.sinks.k1.hdfs.rollSize = 0
    a1.sinks.k1.hdfs.rollCount = 0
    a1.sinks.k1.hdfs.batchSize = 100
    a1.sinks.k1.hdfs.fileType = DataStream
    
    
    #4.2、描述sink2
    a1.sinks.k2.type = hdfs
    a1.sinks.k2.hdfs.path = hdfs://hadoop102:8020/flume/%{ logType}
    a1.sinks.k2.hdfs.filePrefix = user_
    a1.sinks.k2.hdfs.rollInterval = 30
    a1.sinks.k2.hdfs.rollSize = 0
    a1.sinks.k2.hdfs.rollCount = 0
    a1.sinks.k2.hdfs.batchSize = 100
    a1.sinks.k2.hdfs.fileType = DataStream
    
    #5、关联source->channel->sink
    a1.sources.r1.channels = c1 c2
    a1.sinks.k1.channel = c1
    a1.sinks.k2.channel = c2

1.  启动Flume并测试

[hadoop@hadoop102 flume]$ bin/flume-ng agent -c conf/ -n a1 -f job/interceptor.conf -Dflume.root.logger=INFO,console

1.  向9999端口发送数据

[hadoop@hadoop102 ~]$ nc hadoop102 9999
    user:kkkk
    OK
    sys:tttt
    OK

1.  前往HDFS查看数据  
    ![在这里插入图片描述][20201203105047732.png]

[20201203105047732.png]: /images/20221120/6c76be955a204f928041238affc75f85.png