电商离线数仓项目-ETL拦截器/日志类型区分拦截器

小鱼儿 2022-09-13 15:15 174阅读 0赞

Flume里面涉及到拦截器对数据进行处理:  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm6YeMQ29kaW5n_size_20_color_FFFFFF_t_70_g_se_x_16]

本项目中自定义了两个拦截器，分别是：ETL拦截器、日志类型区分拦截器。  
ETL拦截器主要用于，过滤时间戳不合法和json数据不完整的日志  
日志类型区分拦截器主要用于，将错误日志、启动日志和事件日志区分开来，方便发往kafka的不同topic。  
1）创建maven工程flume-interceptor  
2）创建包名：com.atguigu.flume.interceptor  
3）在pom.xml文件中添加如下配置

<dependencies>
        <dependency>
            <groupId>org.apache.flume</groupId>
            <artifactId>flume-ng-core</artifactId>
            <version>1.7.0</version>
        </dependency>
    </dependencies>
    
    <build>
        <plugins>
            <plugin>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>2.3.2</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
            <plugin>
                <artifactId>maven-assembly-plugin</artifactId>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                    <archive>
                        <manifest>
                            <mainClass>com.atguigu.appclient.AppMain</mainClass>
                        </manifest>
                    </archive>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

4）在com.atguigu.flume.interceptor包下创建LogETLInterceptor类名  
Flume ETL拦截器LogETLInterceptor

package com.atguigu.flume.interceptor;
    
    import org.apache.flume.Context;
    import org.apache.flume.Event;
    import org.apache.flume.interceptor.Interceptor;
    import java.nio.charset.Charset;
    import java.util.ArrayList;
    import java.util.List;
    
    public class LogETLInterceptor implements Interceptor { 
    
        @Override
        public void initialize() { 
    
        }
    
        @Override
        public Event intercept(Event event) { 
    
            String body = new String(event.getBody(), Charset.forName("UTF-8"));
    
            // body为原始数据，newBody为处理后的数据,判断是否为display的数据类型
            if (LogUtils.validateReportLog(body)) { 
                return event;
            }
    
            return null;
        }
    
        @Override
        public List<Event> intercept(List<Event> events) { 
    
            ArrayList<Event> intercepts = new ArrayList<>();
    
            // 遍历所有Event，将拦截器校验不合格的过滤掉
            for (Event event : events) { 
                
                Event interceptEvent = intercept(event);
    
                if (interceptEvent != null){ 
                    intercepts.add(interceptEvent);
                }
            }
    
            return intercepts;
        }
    
        @Override
        public void close() { 
    
        }
    
        public static class Builder implements Interceptor.Builder { 
    
            public Interceptor build() { 
                return new LogETLInterceptor();
            }
    
    
            @Override
            public void configure(Context context) { 
    
            }
        }
    }

主要是去执行validateReportLog方法,进行一个合格性的校验  
5）Flume日志过滤工具类

package com.atguigu.flume.interceptor;
    
    import org.apache.commons.lang.StringUtils;
    import org.apache.commons.lang.math.NumberUtils;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    public class LogUtils { 
    
        private static Logger logger = LoggerFactory.getLogger(LogUtils.class);
    
        /**
         * 日志检查，正常的log会返回true，错误的log返回false
         *
         * @param log
         */
        public static boolean validateReportLog(String log) { 
    
            try { 
    //          日志的格式是：时间戳| json串
    //          1549696569054 | { "cm":{ "ln":"-89.2","sv":"V2.0.4","os":"8.2.0","g":"M67B4QYU@gmail.com","nw":"4G","l":"en","vc":"18","hw":"1080*1920","ar":"MX","uid":"u8678","t":"1549679122062","la":"-27.4","md":"sumsung-12","vn":"1.1.3","ba":"Sumsung","sr":"Y"},"ap":"weather","et":[]}
                String[] logArray = log.split("\\|");
    
                if (logArray.length < 2) { 
                    return false;
                }
    //          检查第一串是否为时间戳 或者不是全数字
                if (logArray[0].length() != 13 || !NumberUtils.isDigits(logArray[0])) { 
                    return false;
                }
    
    //          第二串是否为正确的json,这里我们就粗略的检查了，有时候我们需要从后面来发现json传错的数据，做分析
                if (!logArray[1].trim().startsWith("{") || !logArray[1].trim().endsWith("}")) { 
                    return false;
                }
            } catch (Exception e) { 
    //          错误日志打印，需要查看
                logger.error("parse error,message is:" + log);
                logger.error(e.getMessage());
    
                return false;
            }
    
            return true;
        }
    }

**思路:首先是将数据进行切割成时间戳和json两个片段,第一个部分就是对时间戳的校验,该数据是否达到指定的位数或者是该数据里面是否全都都是数字.第二部分对json格式的数据做的一个校验就是首尾去空,之后判断开头和结尾是否以"\{" 和"\}"为开头和结尾的.**

5）Flume日志类型区分拦截器LogTypeInterceptor

package com.atguigu.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;
    
    /**
     * Created by Administrator on 2019/1/18 0018.
     */
    public class LogTypeInterceptor implements Interceptor { 
    
        @Override
        public void initialize() { 
    
        }
    
        @Override
        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);
    
            String logType = "" ;
    
    // startLog
            if (jsonStr.contains("start")) { 
                logType = "start";
            }
            // eventLog
            else { 
                logType = "event";
            }
    
            // 3将日志类型存储到flume头中
            headers.put("logType", logType);
    
            return event;
        }
    
        @Override
        public List<Event> intercept(List<Event> events) { 
    
            ArrayList<Event> interceptors = new ArrayList<>();
    
    for (Event event : events) { 
         Event interceptEvent = intercept(event);
    
         interceptors.add(interceptEvent);
    }
    
    return interceptors;
        }
    
        @Override
        public void close() { 
    
        }
    
        public static class Builder implements Interceptor.Builder { 
    
            public Interceptor build() { 
                return new LogTypeInterceptor();
            }
    
            @Override
            public void configure(Context context) { 
    
            }
        }
    }

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm6YeMQ29kaW5n_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/fbb195cfb4514a14ba82ad5466806fc5.png