flume-ng怎么自定义拦截器

本篇内容主要讲解“flume-ng怎么自定义拦截器”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“flume-ng怎么自定义拦截器”吧!

代码如下:

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

到此，相信大家对“flume-ng怎么自定义拦截器”有了更深的了解，不妨来实际操作一番吧！这里是亿速云网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！