拦截器效果

Flume官方Static拦截器如需设置多组键值：

    
        agent.sources.r.interceptors = i1 i2 i3 agent.sources.r.interceptors.i1.type
        =
        
            static
        
        agent.sources.r.interceptors.i1.key = country agent.sources.r.interceptors.i1.
        
            value
        
        = china agent.sources.r.interceptors.i2.type =
        
            static
        
        agent.sources.r.interceptors.i2.key = province agent.sources.r.interceptors.i2.
        
            value
        
        = fujian agent.sources.r.interceptors.i3.type =
        
            static
        
        agent.sources.r.interceptors.i3.key = city agent.sources.r.interceptors.i3.
        
            value
        
        = xiamen
    

Query拦截器（Query Interceptor）的等效配置：

    
        agent.sources.r.interceptors = i agent.sources.r.interceptors.i.type =
        QueryInterceptor$Builder agent.sources.r.interceptors.i.query = country=china&province=fujian&city=xiamen
    

拦截器配置

创建Maven项目

pom.xml中加入Flume依赖：

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

开发拦截器

自定义拦截器实现 org.apache.flume.interceptor.Interceptor 接口：

    
        
            import
        
        java.util.Map;
        
            import
        
        java.util.List;
        
            import
        
        org.apache.flume.Event;
        
            import
        
        org.apache.flume.Context;
        
            import
        
        org.apache.flume.interceptor.Interceptor;
        
            public
        
        
            
                class
            
            
                QueryInterceptor
            
            
                implements
            
            
                Interceptor
            
        
        {
        
            private
        
        
            final
        
        String query;
        
            //定义key的名称为query
        
        
            public
        
        
            static
        
        
            
                class
            
            
                Constants
            
        
        {
        
            public
        
        
            static
        
        
            final
        
        String KEY =
        
            "query"
        
        ; }
        
            //Builder类（用于Flume获取拦截器实例）
        
        
            public
        
        
            static
        
        
            
                class
            
            
                Builder
            
            
                implements
            
            
                Interceptor
            
            .
            
                Builder
            
        
        {
        
            private
        
        String query;
        
            //返回拦截器实例
        
        
            
                public
            
            Interceptor
            
                build
            
            
                ()
            
        
        {
        
            return
        
        
            new
        
        QueryInterceptor(query); }
        
            //读取Flume配置
        
        
            
                public
            
            
                void
            
            
                configure
            
            
                (Context context)
            
        
        { query = context.getString(Constants.KEY);
        
            //读取query的值
        
        } }
        
            //得到Builder类传来的query值
        
        
            
                private
            
            
                QueryInterceptor
            
            
                (String query)
            
        
        {
        
            this
        
        .query = query; }
        
            //初始化操作（无）
        
        
            
                public
            
            
                void
            
            
                initialize
            
            
                ()
            
        
        { }
        
            //遍历Events并逐个调用intercept方法
        
        
            
                public
            
            List<Event>
            
                intercept
            
            
                (List<Event> events)
            
        
        {
        
            for
        
        (Event e : events) { intercept(e); }
        
            return
        
        events; }
        
            //解析query数据并放入header
        
        
            
                public
            
            Event
            
                intercept
            
            
                (Event event)
            
        
        { Map<String, String> headers = event.getHeaders(); String[] params
        = query.split(
        
            "&"
        
        );
        
            for
        
        (String param : params) { String[] kv = param.split(
        
            "="
        
        );
        
            if
        
        (kv.length ==
        
            2
        
        ) { headers.put(kv[
        
            0
        
        ], kv[
        
            1
        
        ]); } }
        
            return
        
        event; }
        
            //关闭时的操作（无）
        
        
            
                public
            
            
                void
            
            
                close
            
            
                ()
            
        
        { } }
    

启用拦截器

生成Jar包

编译项目生成jar包（依赖不用打包），把jar包放进$FLUME_HOME/plugins.d/QueryInterceptor/lib目录（需创建）

Flume配置

自定义拦截器的type项配置如果有package包名也需写上，并用$分割加上自定义的Builder类名：

    
        
            # source
        
        agent.sources = r agent.sources.r.type =
        
            exec
        
        agent.sources.r.command = tail -F /tmp/test.log agent.sources.r.interceptors
        = i agent.sources.r.interceptors.i.type = QueryInterceptor$Builder agent.sources.r.interceptors.i.query
        = country=china&province=fujian&city=xiamen agent.sources.r.channels
        = c
        
            # channel
        
        agent.channels = c agent.channels.c.type = memory
        
            # sink
        
        agent.sinks = k agent.sinks.k.channel = c agent.sinks.k.type = logger
    

启动Flume

启动Flume时设置-DFlume.root.logger参数：

    
        $FLUME_HOME/bin/flume-ng agent -c
        
            <
            
                Flume配置目录
            
            >
        
        -f
        
            <
            
                配置文件路径
            
            >
        
        -n agent -Dflume.root.logger=INFO,console
    

往/tmp/test.log文件写数据，在控制台可以看到如下信息：

    
        ...
        
            Event
        
        : {
        
            headers
        
        :{country=china, province=fujian, city=xiamen}
        
            body
        
        : ...