第1关：拦截器的使用

任务描述

本关任务：使用拦截器过滤目录下文件中不以y.开头的内容。

编程要求

根据提示，在右侧编辑器补充配置文件，使用拦截器过滤目录下文件中不以y.开头的内容，目录下有如下文件：

    # Define source, channel, sink  
    #agent名称为a1  
    a1.sources = r1  
    a1.channels = c1  
    a1.sinks = k1
    # Define source  
    #source类型配置为avro,监听8888端口，后台会自动发送数据到该端口  
    #拦截后台发送过来的数据，将y.开头的保留下来  
    a1.sources.r1.type = avro  
    a1.sources.r1.channels = c1  
    a1.sources.r1.bind = 0.0.0.0  
    a1.sources.r1.port = 8888  
    a1.sources.r1.interceptors = i1  
    a1.sources.r1.interceptors.i1.type = regex_filter  
    a1.sources.r1.interceptors.i1.regex = y.*  
    #a1.sources.r1.interceptors.i1.excludeEvents =  false
    # Define channel  
    #channel配置为memery  
    a1.channels.c1.type = memory  
    a1.channels.c1.capacity = 100
    # Define sink  
    #落地到 hdfs://localhost:9000/flume目录下  
    #根据时间落地，3s  
    #数据格式DataStream  
    a1.sinks.k1.channel = c1  
    a1.sinks.k1.type = hdfs  
    a1.sinks.k1.hdfs.path = hdfs://localhost:9000/flume  
    a1.sinks.k1.hdfs.rollInterval = 3  
    a1.sinks.k1.hdfs.fileType=DataStream   

第2关：自定义拦截器

任务描述

        本关任务：使用自定义拦截器按照文件名分类采集文件。

相关知识

        Flume 自带拦截器有时候并不能满足我们的业务所需，因此我们需要会自定义拦截器；

编程要求

        目录下有一些命名格式不同的的文件，现要将其按照名称分类采集在hdfs的不同目录，比如ios.log.2018、web.log.2019文件，经过自定义拦截器后采集到hdfs后的目录为：

解题代码

    package com.yy;  
    /**  
     * Licensed to the Apache Software Foundation (ASF) under one  
     * or more contributor license agreements.  See the NOTICE file  
     * distributed with this work for additional information  
     * regarding copyright ownership.  The ASF licenses this file  
     * to you under the Apache License, Version 2.0 (the  
     * "License"); you may not use this file except in compliance  
     * with the License.  You may obtain a copy of the License at  
     *  
     *     http://www.apache.org/licenses/LICENSE-2.0  
     *  
     * Unless required by applicable law or agreed to in writing, software  
     * distributed under the License is distributed on an "AS IS" BASIS,  
     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  
     * See the License for the specific language governing permissions and  
     * limitations under the License.  
     */  
    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;  
    /**  
     * 自定义拦截器  
     * @author yy  
     *  
     */  
    public class RegexExtractorExtInterceptor implements Interceptor {
        static final String REGEX = "regex";  
        static final String SERIALIZERS = "serializers";
        private static final Logger logger = LoggerFactory  
                .getLogger(RegexExtractorExtInterceptor.class);
        private final Pattern regex;  
        private final List<NameAndSerializer> serializers;
        private RegexExtractorExtInterceptor(Pattern regex,  
                                          List<NameAndSerializer> serializers) {  
            this.regex = regex;  
            this.serializers = serializers;  
        }
        @Override  
        public void initialize() {  
            // NO-OP...  
        }
        @Override  
        public void close() {  
            // NO-OP...  
        }
        @Override  
        //拦截单个event  
        public Event intercept(Event event) {
            /**********begin**********/  
            //拦截逻辑
            String Str;  
            //获取header中文件名  
            Str = event.getHeaders().get("basename");  
            //通过正则解析出tomcat.log.2019  
            Matcher matcher = regex.matcher(Str);  
            /**********end**********/
            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  
        //返回Event集合，循环拦截单个Event  
        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 final RegexExtractorInterceptorSerializer defaultSerializer =  
                    new RegexExtractorInterceptorPassThroughSerializer();
            @Override  
            //configure方法  
            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);  
            }
            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  
            //build方法  
            public Interceptor build() {  
                Preconditions.checkArgument(regex != null,  
                        "Regex pattern was misconfigured");  
                Preconditions.checkArgument(serializerList.size() > 0,  
                        "Must supply a valid group match id list");  
                //返回自定义拦截器对象  
                /**********begin**********/  
                return new RegexExtractorExtInterceptor(regex, serializerList);  
                /**********end**********/
            }  
        }
        static class NameAndSerializer {  
            private final String headerName;  
            private final RegexExtractorInterceptorSerializer serializer;
            public NameAndSerializer(String headerName,  
                                     RegexExtractorInterceptorSerializer serializer) {  
                this.headerName = headerName;  
                this.serializer = serializer;  
            }  
        }  
    }  

配置文件：

    # Licensed to the Apache Software Foundation (ASF) under one  
    # or more contributor license agreements.  See the NOTICE file  
    # distributed with this work for additional information  
    # regarding copyright ownership.  The ASF licenses this file  
    # to you under the Apache License, Version 2.0 (the  
    # "License"); you may not use this file except in compliance  
    # with the License.  You may obtain a copy of the License at  
    #  
    #  http://www.apache.org/licenses/LICENSE-2.0  
    #  
    # Unless required by applicable law or agreed to in writing,  
    # software distributed under the License is distributed on an  
    # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY  
    # KIND, either express or implied.  See the License for the  
    # specific language governing permissions and limitations  
    # under the License.
    # The configuration file needs to define the sources,   
    # the channels and the sinks.  
    # Sources, channels and sinks are defined per agent,   
    # in this case called 'agent'
    # Define source, channel, sink  
    #agent名为a1;  
    a1.sources = spool-source1  
    a1.channels = ch1  
    a1.sinks = hdfs-sink1
    # Define and configure an Spool directory source  
    #采集 /opt/flume/data目录下所有文件
    a1.sources.spool-source1.channels = ch1  
    a1.sources.spool-source1.type = spooldir  
    a1.sources.spool-source1.spoolDir = /opt/flume/data  
    a1.sources.spool-source1.fileHeader=true    
    a1.sources.spool-source1.basenameHeader=true   
    a1.sources.spool-source1.interceptors=i1    
    a1.sources.spool-source1.interceptors.i1.type=com.yy.RegexExtractorExtInterceptor$Builder    
    a1.sources.spool-source1.interceptors.i1.regex=(.*)\\.(.*)\\.(.*)  
    a1.sources.spool-source1.interceptors.i1.serializers=s1 s2 s3    
    a1.sources.spool-source1.interceptors.i1.serializers.s1.name=one    
    a1.sources.spool-source1.interceptors.i1.serializers.s2.name=two    
    a1.sources.spool-source1.interceptors.i1.serializers.s3.name=three  
    # Configure channel  
    #channel选择memery  
    a1.channels.ch1.type = memory  
    a1.channels.ch1.capacity = 100
    # Define and configure a hdfs sink  
    #落地到hdfs的hdfs://localhost:9000/flume/文件名的前缀/文件名上的日期  
    #文件格式设为DataStream  
    #根据时间回滚，3s  
    a1.sinks.hdfs-sink1.channel = ch1  
    a1.sinks.hdfs-sink1.type = hdfs  
    a1.sinks.hdfs-sink1.hdfs.path = hdfs://localhost:9000/flume/%{one}/%{three}  
    a1.sinks.hdfs-sink1.hdfs.rollInterval = 3  
    a1.sinks.hdfs-sink1.hdfs.fileType=DataStream