Flume的拦截器和分区使用

墨蓝 2022-06-02 11:37 163阅读 0赞

实际开发中，为了对数据集的方便查询和使用，大型数据集常常被组织为分区（partition）。Flume事件的数据通常按时间来分区。比如把flume监听的数据写入到hdfs中时，就可以进行按时间分区，只需要对hdfs.path的属性进行设置如下：     
                        agent.sinks.sink1.path=/data/flume/year=%Y/month=%m/day=%d

一个Flume事件被写入到哪一个分区是由事件的header中的timestamp(时间戳)决定的，在默认的情况下，事件header中并没有timestamp，但是可以通过Flume的拦截器（interceptor）来添加。拦截器是一种能够对事件流中的事件进行修改或者删除的组件，主要作用是连接source并在事件传递到channel之前对事件进行一次自定义的处理。如下举例给source1增加一个时间戳拦截器，该时间戳拦截器将给source1产生的每个事件添加一个timestamp header:  
                       agent1.sources.souce1.interceptors = interceptor1  
                       agent1.sources.source1.interceptors.interceptor1.type = timestamp。

时间戳拦截器的作用：可以确保这些时间戳能够基本反应事件的创建的时间，然后按照时间戳的时间对其进行分区。但是要注意，时间戳对于单层flume的足够使用，但是如果存在多层Flume代理，那么事件的创建时间和写入时间之间存在明显差异，尤其当代理出现停机的情况时，那么时间戳就不在准确。这种情况下，可以通过设置hdfs.userlocatlTimeStamp属性进行设置，以便使用代理的 时间戳。

实际开发中，很多时候我们要自定义Flume的拦截器，然后打包放到flume安装包下lib包下使用。比如有一次在开发中自定义时间拦截器的使用,是直接修改的TimeStamp类：

package com.misvision.app.flume.interceptor;
    import org.apache.flume.Context;
    import org.apache.flume.Event;
    import org.apache.flume.interceptor.Interceptor;
    import java.util.List;
    import java.util.Map;
    import static org.apache.flume.interceptor.TimestampInterceptor.Constants.*;
    /**
     * 自定义flume的拦截器,提取body中的日志类型作为header
     */
    public class LogCollInterceptor implements Interceptor {
    
        private final boolean preserveExisting;
    
        private LogCollInterceptor(boolean preserveExisting) {
            this.preserveExisting = preserveExisting;
        }
        public void initialize() {
        }
    
        /**
         * Modifies events in-place.
         */
        public Event intercept(Event event) {
            // 1获取flume接收消息头
            Map<String, String> headers = event.getHeaders();
    
            // 2获取flume接收的json数据数组
            byte[] json = event.getBody();
            // 将json数组转换为字符串
            String jsonStr = new String(json);
    
            // pageLog
            String logType = "" ;
            if(jsonStr.contains("pageId")){
                logType = "page" ;
            }
            // eventLog
            else if (jsonStr.contains("eventId")) {
                logType = "event";
            }
            // usageLog
            else if (jsonStr.contains("singleUseDurationSecs")) {
                logType = "usage";
            }
            // error
            else if (jsonStr.contains("errorBrief")) {
                logType = "error";
            }
            // startup
            else if (jsonStr.contains("network")) {
                logType = "startup";
            }
    
            // 3将日志类型存储到flume头中
            headers.put("logType", logType);
    
            return event;
        }
    
        /**
         * Delegates to {@link #intercept(Event)} in a loop.
         * @param events
         * @return
         */
        public List<Event> intercept(List<Event> events) {
            for (Event event : events) {
                intercept(event);
            }
            return events;
        }
    
        public void close() {
        }
    
        public static class Builder implements Interceptor.Builder {
    
            private boolean preserveExisting = PRESERVE_DFLT;
    
            public Interceptor build() {
                return new LogCollInterceptor(preserveExisting);
            }
    
            public void configure(Context context) {
                preserveExisting = context.getBoolean(PRESERVE, PRESERVE_DFLT);
            }
        }
    
        public static class Constants {
            public static String TIMESTAMP = "timestamp";
            public static String PRESERVE = "preserveExisting";
            public static boolean PRESERVE_DFLT = false;
        }
    }
    （3）打成jar包，修改jar包名称为app_logs_flume.jar，然后将该jar包导入linux虚拟机*/flume/lib目录下