文档章节

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

c
 chunhei2008
发布于 2015/03/17 17:03
字数 529
阅读 2764
收藏 0
点赞 0
评论 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

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 总结--flume入门介绍

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

u013362353
05/28
0
0
flume伪分布模式搭建(6)

搭建flume的伪分布的环境 Flume主要解决的是在集群中为hadoop收集数据,一个典型的应用就是web应用,一个大的网站肯定会有好多台服务器的,虽然只有一个中心的数据库,但是为了负载均衡还是会...

lixiyuan
2014/04/04
0
0
flume 拦截器(interceptor)

flume内置了很多拦截器,并且会定期的添加一些拦截器,在这里列出一些flume内置的,经常使用的拦截器。 时间戳拦截器 flume中一个最经常使用的拦截器 ,该拦截器的作用是将时间戳插入到flume...

super_yu
2015/11/17
0
0
Flume NG 简介及配置实战

Flume 作为 cloudera 开发的实时日志收集系统,受到了业界的认可与广泛应用。Flume 初始的发行版本目前被统称为 Flume OG(original generation),属于 cloudera。但随着 FLume 功能的扩展,...

大数据之路
2014/07/08
0
9
Flume 1.3.1 启动时出错解决方案

版本:Flume 1.3.1 启动时出错信息: ./bin/flume-ng: line 81: syntax error in conditional expression: unexpected token ^java.library.path=(.' ./bin/flume-ng: line 81: `if [[ $line......

Hello陈翔
2013/01/09
0
0
两个flume的拦截器(interceptor)

flume支持拦截器(interceptors)机制,是在source这个层面上工作,这里有两个拦截器 1,支持将日志体(event body)里面的字符串替换成另一个字符串。配置文件http://git.oschina.net/atuc...

午火
2014/06/11
0
0
Flume---大数据协作框架

flume是什么 Apache Flume是一个分布式的、可靠的、易用的系统,可以有效地将来自很多不同源系统的大量日志数据收集、汇总或者转移到一个数据中心存储。 Apache Flume的作用不仅限于日志汇总...

简心
05/06
0
0
flume 1.7 源码导入eclipse windows

安装maven,设置MAVEN_HOME等配置 下载flume源码 eclipse-oxygen,设置eclipse 使用外部maven,并配置settings.xml 遇到问题: 如果顺利,已将所需jar都下载下来了。 导入后遇到如下问题 fl...

柯里昂
2017/10/31
0
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

JS中的toFixed,toExponential,toPrecision

toFixed():根据小数点后的指定位数,将数字转换为字符串,它不用指数计数法。 toExponential():根据小数点后指定位数,将数字转化为指数形式字符串。小数点前面只有一位,小数点后面的...

AndyZhouX
9分钟前
0
0
istio 路由实例解析

根据链路追踪图如上: 资料:https://istio.io/docs/guides/bookinfo/ 流程解析: 1. 访问地址: http://IP:31380/productpage kubectl get svc --all-namespaces -o wide istio-system isti......

xiaomin0322
12分钟前
2
0
Centos7通过yum安装jdk8

先查看系统是否已有自带的jdk rpm -qa |grep java rpm -qa |grep jdk rpm -qa |grep gcj 如果没有输出信息,则说明系统没有安装。如果有输出信息,则执行下面的命令卸载 rpm -qa | grep jav...

iplusx
14分钟前
0
0
字体的动画

树不要皮必死无疑,人不要脸天下无敌。如果你此时正在被承受着不公平的待遇,不要伤心不要气馁,吃亏要趁早。 .menu ul li a { position:relative; color: #FFFFFF; text-decoration:...

Js_Mei
15分钟前
0
0
新手学习hadoop发行版本选择介绍

Hadoop对于从事互联网工作的朋友来说已经非常熟悉了,相信在我们身边有很多人正在转行从事hadoop开发的工作,理所当然也会有很多hadoop入门新手。Hadoop开发太过底层,技术难度远比我们想象的...

左手的倒影
15分钟前
0
0
iOS定时器循环引用问题解决

我们通常使用NSTimer或CADisplayLink会使用以下方式 //定义@property (nonatomic, strong)NSTimer *timer;//实现self.timer = [NSTimer scheduledTimerWithTimeInterval:1 target:p......

xiaobai1315
16分钟前
0
0
给wordpress程序提速

在上一节中,我们介绍了wordpress CMS主题提速,本节我们接着介绍wordpress主题提速:gravatar用户头像缓存和google字体去除。 gravatar头像受到全世界网络用户的喜爱,设置好gravatar头像后...

hero2019
18分钟前
0
0
DevExpress v18.1最新版帮助文档下载大全

DevExpress v18.1.4帮助文档下载列表大全来啦!包含.NET、VCL、HTML/JS系列所有帮助文档,提供CHM和PDF两个版本。除已停止更新的Silverlight、Windows 8外,其余均为最新版本。 文章底部扫描...

Miss_Hello_World
22分钟前
0
0
Unity Shader中各种空间及变换方法

前几天尝试写一个传送门的shader,发现自己对坐标之间的变换掌握的不够熟练,趁着这阵子想整理shader相关的知识点,先把各种空间及之间转换整理一下。 1 模型空间-世界空间-观察空间-裁剪空间...

爽歪歪ES
27分钟前
0
0

定义和应用 栈(stack)是一种特殊的线性表,其插入(也称入栈或压栈)和删除(也称出栈或弹栈)操作都在表的同一端进行。这一端被称为栈顶(top)另一端称为栈底端(bottom)。 我们生活中其实...

Frost729
29分钟前
0
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部