手机版
你好,游客 登录 注册 搜索
背景:
阅读新闻

Flume 自定义拦截器

[日期:2018-09-24] 来源:Linux社区  作者:wangpei1949 [字体: ]

做项目时遇到一个问题,需要对接收到的日志数据做复杂逻辑处理并将一条转换成多条。

对比了td-agent,filebeat、flume日志采集工具。

td-agent核心部分是用C实现,而插件部分用了ruby,但ruby不熟;filebeat正则匹配很强大,但关于插件相关资料很少;flume插件却可以直接用Java实现。于是决定通过自定义flume拦截器实现这一功能。

Flume拦截器

Flume的拦截器可删除或修改Event。

Timestamp 拦截器:在Event Header中添加时间戳。
Host 拦截器:在Event Header中添加agent运行机器的Host或IP。
Static 拦截器:在Event Header中添加自定义静态属性。
Remove Header拦截器:可移除Event Header中指定属性。
UUID拦截器:在Event Header中添加全局唯一UUID。
Search and Replace拦截器:基于正则搜索和替换字符串等。
Regex Filtering拦截器:基于正则过滤或反向过滤Event。
Regex Extractor拦截器:基于正则在Event Header添加指定的Key,并将匹配到的内容作为对应的Value。

自定义Flume拦截器

自定义拦截器,实现给每条数据添加公共字段,并将一条转换成多条。其他复杂逻辑类似。

package com.flumePlugins.interceptor;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.codec.Charsets;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.interceptor.Interceptor;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

/**
 * Author: Wang Pei
 * Summary:
 */
public class ParseLogByRule implements Interceptor {
    @Override
    public void initialize() {
        //pass
    }

    @Override
    public void close() {
        //pass

    }

    /**
    * 解析单条event
    * @param event
    * @return
    */
    @Override
    public Event intercept(Event event) {
        //输入
        String inputeBody=null;
        //输出
        byte[] outputBoday=null;
        //解析---这里定义对单条Event处理规则
        try {
            inputeBody=new String(event.getBody(), Charsets.UTF_8);
            ArrayList<String> temp = new ArrayList<>();

            JSONObject bodyObj = JSON.parseObject(inputeBody);

            //1)公共字段
            String host = bodyObj.getString("host");
            String user_id = bodyObj.getString("user_id");
            JSONArray data = bodyObj.getJSONArray("items");

            //2)Json数组=>every json obj
            for (Object item : data) {
                JSONObject itemObj = JSON.parseObject(item.toString());
                HashMap<String, Object> fields = new HashMap<>();
                fields.put("host",host);
                fields.put("user_id",user_id);
                fields.put("item_type",itemObj.getString("item_type"));
                fields.put("active_time",itemObj.getLongValue("active_time"));
                temp.add(new JSONObject(fields).toJSONString());
            }
            //3)Json obj 拼接
            outputBoday=String.join("\n",temp).getBytes();
        }catch (Exception e){
            System.out.println("输入数据:"+inputeBody);
            e.printStackTrace();
        }
        event.setBody(outputBoday);
        return event;
    }

    /**
    * 解析一批event
    * @param events
    * @return
    */
    @Override
    public List<Event> intercept(List<Event> events) {
        //输出---一批Event
        ArrayList<Event> result = new ArrayList<>();
        //输入---一批Event
        try{
            for (Event event : events) {
                //一条条解析
                Event interceptedEvent = intercept(event);
                byte[] interceptedEventBody = interceptedEvent.getBody();
                if(interceptedEventBody.length!=0){
                    String multiEvent = new String(interceptedEventBody, Charsets.UTF_8);
                    String[] multiEventArr = multiEvent.split("\n");
                    for (String needEvent : multiEventArr) {
                        SimpleEvent simpleEvent = new SimpleEvent();
                        simpleEvent.setBody(needEvent.getBytes());
                        result.add(simpleEvent);
                    }

                }
            }

        }catch (Exception e){
            e.printStackTrace();
        }
        return result;
    }

    /**
    * 实现内部类接口
    */
    public static class Builder implements Interceptor.Builder{
        @Override
        public Interceptor build() {
            return new ParseLogByRule();
        }

        @Override
        public void configure(Context context) {

        }
    }

}

插件打包上传

编译打包拦截器插件,然后将打包后的插件和依赖的fastjson一起上传到flume lib目录

配置agent

TAILDIR Source ===>file Channel ===>Kafka Sink

    cat agent.conf
    # source的名字
    agent.sources = s1
    # channels的名字
    agent.channels = c1
    # sink的名字
    agent.sinks = r1

    # 指定source使用的channel
    agent.sources.s1.channels = c1
    # 指定sink使用的channel
    agent.sinks.r1.channel = c1

    ######## source相关配置 ########
    # source类型
    agent.sources.s1.type = TAILDIR
    # 元数据位置
    agent.sources.s1.positionFile = /Users/wangpei/tempData/flume/taildir_position.json
    # 监控的目录
    agent.sources.s1.filegroups = f1
    agent.sources.s1.filegroups.f1=/Users/wangpei/tempData/flume/data/.*log
    agent.sources.s1.fileHeader = true

    ######## interceptor相关配置 ########
    agent.sources.s1.interceptors = i1
    agent.sources.s1.interceptors.i1.type = com.flumePlugins.interceptor.ParseLogByRule$Builder

    ######## channel相关配置 ########
    # channel类型
    agent.channels.c1.type = file
    # 数据存放路径
    agent.channels.c1.dataDirs = /Users/wangpei/tempData/flume/filechannle/dataDirs
    # 检查点路径
    agent.channels.c1.checkpointDir = /Users/wangpei/tempData/flume/filechannle/checkpointDir
    # channel中最多缓存多少
    agent.channels.c1.capacity = 1000
    # channel一次最多吐给sink多少
    agent.channels.c1.transactionCapacity = 100

    ######## sink相关配置 ########
    # sink类型
    agent.sinks.r1.type = org.apache.flume.sink.kafka.KafkaSink
    # brokers地址
    agent.sinks.r1.kafka.bootstrap.servers = localhost:9092
    # topic
    agent.sinks.r1.kafka.topic = testTopic3
    # 压缩
    agent.sinks.r1.kafka.producer.compression.type = snappy

启动flume-agent

bin/flume-ng agent --conf conf/ -f conf/agent.conf -Dflume.root.logger=INFO,console -name agent

启动kafka-console-consumer

./kafka-console-consumer --topic testTopic3 --bootstrap-server localhost:9092

写入测试数据到监控目录

#日志数据
log='{
"host":"www.baidu.com",
"user_id":"197878787878787",
"items":[
    {
        "item_type":"clothes",
        "active_time":18989989
    },
    {
        "item_type":"car",
        "active_time":18989989
    }
 ]
}'
#日志追加到文件
echo $log>> /Users/wangpei/tempData/flume/data/test.log

查看发送到kafka中的数据

#可以看到一条数据按规则被解析成了两条
{"active_time":18989989,"user_id":"197878787878787","item_type":"clothes","host":"www.baidu.com"}
{"active_time":18989989,"user_id":"197878787878787","item_type":"car","host":"www.baidu.com"}

自定义拦截器要点

A、实现Interceptor接口中intercept(Event event)方法和intercept(List<Event> events)方法。

B、创建内��类Builder实现Interceptor.Builder接口。

C、注意对异常数据的处理。防止Agent奔溃。

总结

通过拦截器确实实现了这一功能,为解决这一类问题提供了一种很好的思路,但逻辑太过复杂反而会降低flume同步效率,生产环境下还要多加验证。

Linux公社的RSS地址https://www.linuxidc.com/rssFeed.aspx

本文永久更新链接地址https://www.linuxidc.com/Linux/2018-09/154359.htm

linux
相关资讯       Flume 
本文评论   查看全部评论 (0)
表情: 表情 姓名: 字数

       

评论声明
  • 尊重网上道德,遵守中华人民共和国的各项有关法律法规
  • 承担一切因您的行为而直接或间接导致的民事或刑事法律责任
  • 本站管理人员有权保留或删除其管辖留言中的任意内容
  • 本站有权在网站内转载或引用您的评论
  • 参与本评论即表明您已经阅读并接受上述条款