文档章节

flume-ng 自定义拦截器,对header中的字段进行正则匹配分离出更多header

c
 chunhei2008
发布于 2015/03/17 17:03
字数 529
阅读 2782
收藏 0


代码如下:

package com.wy.flume.interceptor;

import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.lang.StringUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import org.apache.flume.interceptor.RegexExtractorInterceptorPassThroughSerializer;
import org.apache.flume.interceptor.RegexExtractorInterceptorSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Charsets;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import com.google.common.collect.Lists;

public class RegexExtractorHeaderInterceptor implements Interceptor {

    static final String REGEX = "regex";
    static final String SERIALIZERS = "serializers";
    
    
    static final String EXTRACTOR_HEADER = "extractorHeader";  
    static final boolean DEFAULT_EXTRACTOR_HEADER = false;  
    static final String EXTRACTOR_HEADER_KEY = "extractorHeaderKey"; 

    private static final Logger logger = LoggerFactory
        .getLogger(RegexExtractorHeaderInterceptor.class);

    private final Pattern regex;
    private final List<NameAndSerializer> serializers;

    private final boolean extractorHeader;  
    private final String extractorHeaderKey;  
    
    private RegexExtractorHeaderInterceptor(Pattern regex,
        List<NameAndSerializer> serializers,boolean extractorHeader, String extractorHeaderKey) {
      this.regex = regex;
      this.serializers = serializers;
      
      this.extractorHeader = extractorHeader;
      this.extractorHeaderKey = extractorHeaderKey;
      
    }

    @Override
    public void initialize() {
      // NO-OP...
    }

    @Override
    public void close() {
      // NO-OP...
    }

    @Override
    public Event intercept(Event event) {
      String extractorHeaderVal;
      if (extractorHeader){
          
          extractorHeaderVal = event.getHeaders().get(extractorHeaderKey);
          
      }else{
          
          extractorHeaderVal = new String(event.getBody(),Charsets.UTF_8);
          
      }
      
      Matcher matcher = regex.matcher(extractorHeaderVal);
      Map<String, String> headers = event.getHeaders();
      if (matcher.find()) {
        for (int group = 0, count = matcher.groupCount(); group < count; group++) {
          int groupIndex = group + 1;
          if (groupIndex > serializers.size()) {
            if (logger.isDebugEnabled()) {
              logger.debug("Skipping group {} to {} due to missing serializer",
                  group, count);
            }
            break;
          }
          NameAndSerializer serializer = serializers.get(group);
          if (logger.isDebugEnabled()) {
            logger.debug("Serializing {} using {}", serializer.headerName,
                serializer.serializer);
          }
          headers.put(serializer.headerName,
              serializer.serializer.serialize(matcher.group(groupIndex)));
        }
      }
      return event;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
      List<Event> intercepted = Lists.newArrayListWithCapacity(events.size());
      for (Event event : events) {
        Event interceptedEvent = intercept(event);
        if (interceptedEvent != null) {
          intercepted.add(interceptedEvent);
        }
      }
      return intercepted;
    }

    public static class Builder implements Interceptor.Builder {

      private Pattern regex;
      private List<NameAndSerializer> serializerList;
      
      private boolean extractorHeader;
      private String extractorHeaderKey;
      
      private final RegexExtractorInterceptorPassThroughSerializer defaultSerializer = new RegexExtractorInterceptorPassThroughSerializer();
      

      @Override
      public void configure(Context context) {
        String regexString = context.getString(REGEX);
        Preconditions.checkArgument(!StringUtils.isEmpty(regexString),
            "Must supply a valid regex string");
        regex = Pattern.compile(regexString);
        regex.pattern();
        regex.matcher("").groupCount();
        configureSerializers(context);
        
        extractorHeader = context.getBoolean(EXTRACTOR_HEADER,DEFAULT_EXTRACTOR_HEADER);
        
        if (extractorHeader){
            
            extractorHeaderKey = context.getString(EXTRACTOR_HEADER_KEY);
            Preconditions.checkArgument(!StringUtils.isEmpty(extractorHeaderKey),"header key must");
            
        }
        
      }

      private void configureSerializers(Context context) {
        String serializerListStr = context.getString(SERIALIZERS);
        Preconditions.checkArgument(!StringUtils.isEmpty(serializerListStr),
            "Must supply at least one name and serializer");

        String[] serializerNames = serializerListStr.split("\\s+");

        Context serializerContexts =
            new Context(context.getSubProperties(SERIALIZERS + "."));

        serializerList = Lists.newArrayListWithCapacity(serializerNames.length);
        for(String serializerName : serializerNames) {
          Context serializerContext = new Context(
              serializerContexts.getSubProperties(serializerName + "."));
          String type = serializerContext.getString("type", "DEFAULT");
          String name = serializerContext.getString("name");
          Preconditions.checkArgument(!StringUtils.isEmpty(name),
              "Supplied name cannot be empty.");

          if("DEFAULT".equals(type)) {
            serializerList.add(new NameAndSerializer(name, defaultSerializer));
          } else {
            serializerList.add(new NameAndSerializer(name, getCustomSerializer(
                type, serializerContext)));
          }
        }
      }

      private RegexExtractorInterceptorSerializer getCustomSerializer(
          String clazzName, Context context) {
        try {
          RegexExtractorInterceptorSerializer serializer = (RegexExtractorInterceptorSerializer) Class
              .forName(clazzName).newInstance();
          serializer.configure(context);
          return serializer;
        } catch (Exception e) {
          logger.error("Could not instantiate event serializer.", e);
          Throwables.propagate(e);
        }
        return defaultSerializer;
      }

      @Override
      public Interceptor build() {
        Preconditions.checkArgument(regex != null,
            "Regex pattern was misconfigured");
        Preconditions.checkArgument(serializerList.size() > 0,
            "Must supply a valid group match id list");
        return new RegexExtractorHeaderInterceptor(regex, serializerList, extractorHeader, extractorHeaderKey);
      }
    }

    static class NameAndSerializer {
      private final String headerName;
      private final RegexExtractorInterceptorSerializer serializer;

      public NameAndSerializer(String headerName,
          RegexExtractorInterceptorSerializer serializer) {
        this.headerName = headerName;
        this.serializer = serializer;
      }
    }
  }

应用配置:

hdp2.sources.s1.interceptors = i2
hdp2.sources.s1.interceptors.i2.type = com.wy.flume.interceptor.RegexExtractorHeaderInterceptor$Builder
hdp2.sources.s1.interceptors.i2.regex = ([^_]+)_(\\d{8}).*
hdp2.sources.s1.interceptors.i2.extractorHeader = true
hdp2.sources.s1.interceptors.i2.extractorHeaderKey = basename
hdp2.sources.s1.interceptors.i2.serializers = s1 s2
hdp2.sources.s1.interceptors.i2.serializers.s1.name = log_type
hdp2.sources.s1.interceptors.i2.serializers.s2.name = log_day

© 著作权归作者所有

共有 人打赏支持
c
粉丝 3
博文 27
码字总数 8480
作品 2
广州
高级程序员
Flume学习系列(四)---- Interceptors(拦截器)

前言:flume通过使用Interceptors(拦截器)实现修改和过滤事件的功能。举个栗子,一个网站每天产生海量数据,但是可能会有很多数据是不完整的(缺少重要字段),或冗余的,如果不对这些数据...

小北觅
08/21
0
0
阿里大数据工程师教你怎样理解Flume

lume是干什么的? 收集日志的 flume如何搜集日志? 我们把flume比作情报人员 (1)搜集信息 (2)获取记忆信息 (3)传递报告间谍信息 flume是怎么完成上面三件事情的,三个组件: source: ...

JAVA丶学习
04/14
0
0
Flume NG 学习笔记(八)Interceptors(拦截器)测试

版权声明:本文为博主原创文章,未经博主允许不得转载。 目录(?)[+] 拦截器主要是对事件的header信息信息操作,要么直接忽略他,要么修改他的数据 一、Event Serializers file_roll sink 和h...

jackwxh
06/29
0
0
Flume学习系列(五)---- Custom Interceptors(自定义拦截器)

前言:接上一篇,本篇文章实现一个自定义的拦截器。主要功能是在Event的body中添加IP地址。因为没有拦截器可以在Body中添加(host是在header中添加),所以需要自定义。掌握了这个,其他的情...

小北觅
08/21
0
0
flume 总结--flume入门介绍

flume介绍 flume被设计为一个灵活的分布式系统,可以很容易的扩展,而且是高度可定制化的,一个配置正确的Flume Agent和由互相连接的Agent创建的Agent管道,保证不会丢失数据,提供持久的cha...

u013362353
05/28
0
0

没有更多内容

加载失败,请刷新页面

加载更多

在Debian 9系统上安装Mysql数据库的方法教程

前言 看到题目大家应都会想,在 Debian 9 上安装 Mysql?那不是很简单的事儿吗?直接 sudo apt install mysql-server 不就行了吗? 没想到遇到了几个之前没遇到的问题,耽误了不少时间。 原来...

临江仙卜算子
41分钟前
2
0
从web实时通信讲H5 WebSocket

通常我们打开一个浏览器访问网页时,都会向页面所在的服务器发送一个HTTP请求,然后web服务器确认请求并向浏览器做出响应。简单的说,就是一个请求对应的一个响应。然而这种方法对许多的应用...

Code辉
54分钟前
3
0
Sharding-Sphere自动化执行引擎

Q: 什么叫"自动化执行引擎"? A: 一条SQL的生命周期是:从客户端发起、经过Sharding-Sphere处理、再到底层数据库执行消化。而在Sharding-Sphere里过程则是:SQL解析-->SQL优化-->SQL路由-->...

xiaomin0322
57分钟前
2
0
单模块中ReentrantLock的使用

背景 在单模块应用中,对同一个请求,需要进行同步。注意ReentrantLock的使用场景: 同一个线程中 同一个请求 RestController @RestControllerpublic class Controller {private final Re...

亚林瓜子
59分钟前
2
0
Linux 4.1内核热补丁成功实践

好久不见的干货重现江湖!今日的内容是基于UCloud运维同学反馈的个别宿主机上存在进程CPU峰值使用率异常现象问题进行的相关阐述。本文详细介绍了该问题的完整分析思路和用热补丁的方式成功解...

UCloudTech
今天
1
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部