共计 6411 个字符，预计需要花费 17 分钟才能阅读完成。
本篇内容主要讲解“flume-ng 怎么自定义拦截器”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让丸趣 TV 小编来带大家学习“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 怎么自定义拦截器”有了更深的了解，不妨来实际操作一番吧！这里是丸趣 TV 网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！
正文完
