Flume 自定義攔截器 多行讀取日志+截斷

前言：

　　Flume百度定義如下：

       Flume是Cloudera提供的一個高可用的，高可靠的，分布式的海量日志采集、聚合和傳輸的系統，Flume支持在日志系統中定制各類數據發送方，用於收集數據；同時，Flume提供對數據進行簡單處理，並寫到各種數據接受方（可定制）的能力。

搭建並使用flume不是特別難，而且網上也有技術文章分享，我不再贅述了。本文主要建立在已經搭建並使用flume的情況。

 

業務場景：

flume讀取日志是按行讀取，無法進行多行讀取，當出現如下日志時將無法讀到日志的正確時間與類型信息,所以我們需要有一種可以多行讀取日志信息的辦法,這里采用自定義攔截器的方法：

1 2019-08-02 14:34:13.153 [DEBUG][tomcatThreadPool-7][com.xxx.xxx.xxx.xxx.web.CommonHandlerExceptionResolver] (CommonHandlerExceptionResolver.java:134) \n Exception:------------------------------------------------------------------\ncom.xxx.xxx.xx.exceptions.XxxException: \n### Error querying database.  Cause: com.mysql.jdbc.PacketTooBigException: Packet for query is too large (5638500 > 4194304). You can change this value on the server by setting the max_allowed_packet' variable.\n### The error may exist in com/xxx/xxx/xxx/basic/mapper/custom/CsWmFrtRateExtMapper.xml\n### The error may involve defaultParameterMap\n### The error occurred while setting parameters\n### SQL: SELECT * FROM ((SELECT cwfr.CS_WM_FRT_RATE_ID AS CS_WM_FRT_RATE_ID, cwfr.ACTIVE_DATE_BEGIN AS ACTIVE_DATE_BEGIN, cwfr.FRT_ITEM_CODE AS FRT_ITEM_CODE, cwfr.FRT_ITEM_NAME AS FRT_ITEM_NAME, cwfr.RP_FLAG AS RP_FLAG, cwfr.FRT_MODE AS FRT_MODE, cwfr.CALCULATION_ITEM AS CALCULATION_ITEM, cwfr.CHARGE_UOM_CODE AS CHARGE_UOM_CODE, cwfr.CHARGE_UOM_NAME AS 

  

對於一些業務系統的日志可能會比較大，超1M，2M甚至更多，可以根據實際情況只截取前面一部分保留下來即可，為了讓功能更具有靈活性，在實現上增加開關屬性，默認打開着，不需要時設置關閉。

自定義攔截器實現的屬性：過濾正則，截斷標識(即開關)，總截取最大長度，單個截取最大長度，最后一個事件流。最后一個事件流的作用保留下來與下一批次一起，按正則匹配后才發送出去，因為flume是按批次讀取的，默認是100行，而這個配置又與flume運行內存有關系。這個是屬於參數調優的話題。

 

特別注意：代碼打包后是需要放到flume安裝目錄下的lib下。放進去后需要重新才會生效。

 

代碼實現如下：

package org.apache.flume.custom;

import com.google.common.collect.Lists;
import org.apache.commons.codec.Charsets;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * 自定義攔截器 參考 Author: xiufen.huang Create Data: 2019/8/12 15:46
 */
public class MultInterceptor implements Interceptor {

    // 過濾正則
    private static Pattern regex = null;
    // 截取標志
    private static Boolean cutFlag = true;
    // 總截取最大長度
    private static Integer cutMax = null;
    // 單個截取最大長度
    private static Integer singleCut = null;
    // 最后一個事件流
    private static List<Event> lastList = Lists.newArrayList();

    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {
//        System.out.println("----------intercept(Event event)方法執行，處理單個event");  
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> list) {
//        System.out.println("進來方法了嗎？");

        // 處理結果 event list
        List<Event> intercepted = null;

        int addnum = 0;// 記錄上一個正確匹配的event在隊列中的位置,以便下一event有和它連接的需要

        if (lastList != null && lastList.size() >0){
            // 初始化
            int initCapacity = list.size() + lastList.size();
            intercepted = Lists.newArrayListWithCapacity(initCapacity);
            // 添加
            intercepted.addAll(lastList);

            // 清空
            lastList = Lists.newArrayList();
        }else {
            intercepted = Lists.newArrayListWithCapacity(list.size());
        }

        // 有正則的情況
        for (int i = 0; i < list.size(); i++) {
            Event interceptedEvent  = null;
            Matcher matcher = regex.matcher(new String(list.get(i).getBody(), Charsets.UTF_8));
            if (matcher.find()) {
                interceptedEvent = intercept((Event)list.get(i));
                // 單個的body
                String singleBody = new String(interceptedEvent.getBody(), Charsets.UTF_8);
                int singleBodyLen = singleBody.length();
                System.out.println("正則匹配-原始body---------:" + singleBody);
                if (cutFlag) {
                    // 處理最大截取數邊界條件--一定要重新一個變量接收
                    int lsSingleCut = singleCut > singleBodyLen ? singleBodyLen : singleCut;
                    // 截取字符串--新變量
                    String singleCutBody = new String(singleBody.substring(0, lsSingleCut));

                    System.out.println("單個截取-截取后body=============:" + singleCutBody);
                    // 重新賦值body
                    interceptedEvent.setBody(singleCutBody.getBytes());
                }

                intercepted.add(interceptedEvent);
                addnum = addnum +1;
//                System.out.println("matcher.find() 下的：addnum：" + addnum);
            } else {
                if (intercepted.size() == 0) {
                    // 表示本次沒有匹配上
                    continue;
                }

                 addnum = addnum >= intercepted.size() ? intercepted.size() - 1 : addnum;


                String body = new String(intercepted.get(addnum).getBody(), Charsets.UTF_8) + "\n"
                        + new String(list.get(i).getBody(), Charsets.UTF_8);

                System.out.println("總截取-原始body---------:" + body);
                int bodyLen = body.length();
                // 截取body-新變量
                String cutBody = body;
                if (cutFlag) {

                    // 處理最大截取數邊界條件--新變量
                    int lsCutMax = cutMax > bodyLen ? bodyLen : cutMax;
                    // 截取字符串
                    cutBody = new String(body.substring(0, lsCutMax));
                    System.out.println("-處理截取-截取后body=============: " + body);
                }

                intercepted.get(addnum).setBody(cutBody.getBytes());
            }
        }

        // 最后一個保存在靜態變量，等待下一批次
        if (intercepted != null && intercepted.size() > 0){
            int lastIndex = intercepted.size() -1;
            lastList.add(intercepted.get(lastIndex));
            // 移除最后一個索引
            intercepted.remove(lastIndex);
        }

        return intercepted;
    }

    @Override
    public void close() {
        System.out.println("----------自定義攔截器close方法執行");
    }

    public static class Builder implements Interceptor.Builder {
        @Override
        public Interceptor build() {
            System.out.println("----------build方法執行");
            return new MultInterceptor();
        }

        @Override
        public void configure(Context context) {
            String regexStr = context.getString("regex", null);
            cutFlag = context.getBoolean("cutFlag", true);
            cutMax = context.getInteger("cutMax", 0);
            singleCut = context.getInteger("singleCut", 0);
            System.out.println("參數regexStr：" + regexStr + ",參數cutMax： " + cutMax + ",cutFlag: " + cutFlag
                    + " ,singleCut: " + singleCut);

            // 由於外面傳過來的單位是kb，所以這邊需要乘以1024
            cutMax = cutMax * 1024;
            System.out.println("總截取最大值：" + cutMax);
            singleCut = singleCut * 1024;
            System.out.println("單個截取最大值：" + singleCut);

            if (null != regexStr) {
                // 轉換正則
                regex = Pattern.compile(regexStr);
            }

        }
    }
}

 

使用說明：

在flume啟動配置文件增加以下內容：

#匹配時間並轉換為時間戳到header中
a1.sources.tail.interceptors.i2.type=org.apache.flume.custom.MultInterceptor$Builder
#正則表達式，按需求定
a1.sources.tail.interceptors.i2.regex=(((?!0000)[0-9]{4}-((0[1-9]|1[0-2])-(0[1-9]|1[0-9]|2[0-8])|(0[13-9]|1[0-2])-(29|30)|(0[13578]|1[02])-31)|([0-9]{2}(0[48]|[2468][048]|[13579][26])|(0[48]|[2468][048]|[13579][26])00)-02-29))
#開啟日志長度截取標志，默認true，開啟
a1.sources.tail.interceptors.i2.cutFlag = true
#最大截取字符串長度,整數,盡量控制在2M以內，單位：kb，1M=1024
a1.sources.tail.interceptors.i2.cutMax = 2048
#單個截取字符串長度，整數，盡量控制在1.5M以內，單位：kb,1M=1024
a1.sources.tail.interceptors.i2.singleCut=1024
a1.sources.tail.interceptors.i2.serializers=se1
a1.sources.tail.interceptors.i2.serializers.se1.type=org.apache.flume.interceptor.RegexExtractorInterceptorMillisSerializer
a1.sources.tail.interceptors.i2.serializers.se1.name=timestamp
a1.sources.tail.interceptors.i2.serializers.se1.pattern=yyyy-MM-dd

 

 

參考實現：

flume 自定義攔截器實現多行讀取日志 https://blog.csdn.net/nougats/article/details/71188920