最码农 最码农
  • 首页
  • 动态广场
  • 精选栏目
  • 闲言碎语
  • 左邻右里
  • 笔记屋
  • 注册
  • 登录
首页 › 大数据 › Flume 自定义Interceptor

Flume 自定义Interceptor

Cosy
1月前大数据阅读 92

目录

Flume

自定义Interceptor

1)案例需求

使用Flume采集服务器本地日志,需要按照日志类型的不同,将不同种类的日志发往不同的分析系统。

2)需求分析

在实际的开发中,一台服务器产生的日志类型可能有很多种,不同类型的日志可能需要发送到不同的分析系统。此时会用到Flume拓扑结构中的Multiplexing结构,Multiplexing的原理是,根据event中Header的某个key的值,将不同的event发送到不同的Channel中,所以我们需要自定义一个Interceptor,为不同类型的event的Header中的key赋予不同的值。

在该案例中,我们以端口数据模拟日志,以数字(单个)和字母(单个)模拟不同类型的日志,我们需要自定义interceptor区分数字和字母,将其分别发往不同的分析系统(Channel)。

Flume 自定义Interceptor-最码农

3)实现步骤

(1)创建一个maven项目,并引入以下依赖。

<dependency>
    <groupId>org.apache.flume</groupId>
    <artifactId>flume-ng-core</artifactId>
    <version>1.9.0</version>
</dependency>

(2)定义CustomInterceptor类并实现Interceptor接口。


package com.cosyblogs.interceptor;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.List;

public class CustomInterceptor implements Interceptor {
    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {
        byte[] body = event.getBody();
        if (body[0] < 'z' && body[0] > 'a') {
            event.getHeaders().put("type", "letter");
        } else if (body[0] > '0' && body[0] < '9') {
            event.getHeaders().put("type", "number");
        }
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> list) {
        for (Event event : list) {
            intercept(event);
        }
        return list;

    }

    @Override
    public void close() {

    }

    public static class Builder implements Interceptor.Builder {

        @Override
        public Interceptor build() {
            return new CustomInterceptor();
        }

        @Override
        public void configure(Context context) {
        }

    }

}

(3)编辑flume配置文件。

为hadoop102上的Flume1配置1个netcat source,1个sink group(2个avro sink),并配置相应的ChannelSelector和interceptor。


# Name the components on this agent
a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1 c2

# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 6666
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.cosyblogs.interceptor.CustomInterceptor$Builder
a1.sources.r1.selector.type = multiplexing
a1.sources.r1.selector.header = type
a1.sources.r1.selector.mapping.letter = c1
a1.sources.r1.selector.mapping.number = c2
# Describe the sink
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = hadoop103
a1.sinks.k1.port = 4141

a1.sinks.k2.type=avro
a1.sinks.k2.hostname = hadoop104
a1.sinks.k2.port = 4242

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Use a channel which buffers events in memory
a1.channels.c2.type = memory
a1.channels.c2.capacity = 1000
a1.channels.c2.transactionCapacity = 100


# Bind the source and sink to the channel
a1.sources.r1.channels = c1 c2
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c2

为hadoop103上的Flume2配置一个avro source和一个logger sink。


a1.sources = r1
a1.sinks = k1
a1.channels = c1

a1.sources.r1.type = avro
a1.sources.r1.bind = hadoop103
a1.sources.r1.port = 4141

a1.sinks.k1.type = logger

a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

a1.sinks.k1.channel = c1
a1.sources.r1.channels = c1

为hadoop104上的Flume3配置一个avro source和一个logger sink。


a1.sources = r1
a1.sinks = k1
a1.channels = c1

a1.sources.r1.type = avro
a1.sources.r1.bind = hadoop104
a1.sources.r1.port = 4242

a1.sinks.k1.type = logger

a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

a1.sinks.k1.channel = c1
a1.sources.r1.channels = c1

(4)分别在hadoop102,hadoop103,hadoop104上启动flume进程,注意先后顺序。

(5)在hadoop102使用netcat向localhost:6666发送字母和数字。


[hadoop@hadoop102 ~]$ nc localhost 6666

hello
OK
666
OK

(6)观察hadoop103和hadoop104打印的日志。

hadoop103

Flume 自定义Interceptor-最码农

hadoop104

Flume 自定义Interceptor-最码农
Flume 教程
赞(0) 收藏(0)
Flume 进阶案例 – 聚合
上一篇
Flume 自定义Source
下一篇
再想想
暂无评论
随 机 推 荐
Hive 行转列与列转行
Hive 常用函数整理
Flume 自定义Sink
Flume 入门案例 – 实时监控单个追加文件
从Hadoop框架讨论大数据生态
HDFS 概述
MapReduce框架原理-InputFormat数据输入
Hadoop 序列化
  • 0
  • 0
介绍

我们致力于打造一个原创的计算机相关技术的博客网站,旨在为访客提供一个优质的计算机技术教程交流平台。网站开辟了很多于计算机相关的栏目,并且收集了不少实用资源,同时也鼓励欢迎访客一起分享、交流、学习。

灵魂推荐
Veer图库 数码荔枝
栏目标题
首页 动态广场 精选栏目 闲言碎语 左邻右里 笔记屋
Copyright © 2021-2023 最码农. 苏ICP备20033168号
  • 首页
  • 动态广场
  • 精选栏目
  • 闲言碎语
  • 左邻右里
  • 笔记屋
# 教程 # # Hadoop # # Hive # # Flume # # 人工智能 #
Cosy
即使世界毁灭,也总有回光返照的那一刻
117
文章
3
评论
432
喜欢