Flume自定义组件

flume source种类

flume官方支持很多源，比如：netcat source，avro source，Kafka source，exec source，taildir source或者自定义source等等。

source文档

flume官方支持多种目标源，比如：hdfs sink，hive sink，logger sink，kafka sink等等，同样也支持自定义sink。

sink文档

自定义拦截器（Interceptor）

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

1. 创建一个 Maven 项目，并引入以下依赖。

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

1. 定义 CustomInterceptor 类并实现 Interceptor 接口。之后将 Maven 项目打成 jar 包上传到 /flume/lib 目录下。

package com.neu.interceptor;

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

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public class CustomInterceptor implements Interceptor {

    // 声明一个存放事件的集合
    private List<Event> addHeaderEvents;

    public void initialize() {
        addHeaderEvents = new ArrayList<Event>();
    }

    // 单个事件拦截
    public Event intercept(Event event) {
        // 1.获取事件中的头信息
        Map<String, String> headers = event.getHeaders();
        // 2.获取事件中的body信息
        String body = new String(event.getBody());
        // 3.根据body中是否有“hello”来决定添加怎样的头信息
        if (body.contains("hello")) {
            headers.put("type", "neu");
        } else {
            headers.put("type", "others");
        }
        return event;
    }

    // 批量事件拦截
    public List<Event> intercept(List<Event> events) {
        // 1.清空集合
        addHeaderEvents.clear();
        // 2.遍历events
        for (Event event : events) {
            // 3.为每个事件添加头信息
            addHeaderEvents.add(intercept(event));
        }
        return addHeaderEvents;
    }

    public void close() {

    }

    public static class Builder implements Interceptor.Builder {

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

        public void configure(Context context) {

        }
    }
}

1. 创建 flume1.conf

   配置一个 Netcat Source，一个 Sink Group（2 个 Avro Sink）,并配置相应 ChannelSelector 和 Interceptor。

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

# Interceptor
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.neu.interceptor.CustomInterceptor$Builder

# Channel Selector
a1.sources.r1.selector.type = multiplexing
a1.sources.r1.selector.header = type
a1.sources.r1.selector.mapping.neu = c1
a1.sources.r1.selector.mapping.others = c2

# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444

# Describe the sink
# sink 端的 avro 是一个数据发送者
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = master
a1.sinks.k1.port = 4141
a1.sinks.k2.type = avro
a1.sinks.k2.hostname = slave2
a1.sinks.k2.port = 4142

# Describe the channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
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

1. 创建 flume2.conf

配置一个 Avro Source 和 logger Sink。

# Name the components on this agent
a2.sources = r1
a2.sinks = k1
a2.channels = c1

# Describe/configure the source
a2.sources.r1.type = avro
a2.sources.r1.bind = slave2
a2.sources.r1.port = 4142

# Describe the sink
a2.sinks.k1.type = logger

# Describe the channel
a2.channels.c1.type = memory
a2.channels.c1.capacity = 1000
a2.channels.c1.transactionCapacity = 100

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

1. 创建 flume3.conf

配置一个 Avro Source 和 logger Sink。

# Name the components on this agent
a3.sources = r1
a3.sinks = k1
a3.channels = c1

# Describe/configure the source
a3.sources.r1.type = avro
a3.sources.r1.bind = master
a3.sources.r1.port = 4141

# Describe the sink
a3.sinks.k1.type = logger

# Describe the channel
a3.channels.c1.type = memory
a3.channels.c1.capacity = 1000
a3.channels.c1.transactionCapacity = 100

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

1. 开启监控

bin/flume-ng agent -c conf -f job/interceptor/flume1.conf -n a1

bin/flume-ng agent -c conf -f job/interceptor/flume2.conf -n a2 -Dflume.root.logger=INFO,console

bin/flume-ng agent -c conf/ -f job/interceptor/flume3.conf -n a3 -Dflume.root.logger=INFO,console

自定义 Source

  使用 flume 接收数据，并给每条数据添加前缀，输出到控制台。前缀可以从 flume 配置文件中配置。

1. 定义 MySource 类，继承 AbstractSource 类并实现 Configurable 和 PollableSource 接口。之后将 Maven 项目打成 jar 包上传到 /flume/lib 目录下。

package com.neu.interceptor.source;

import org.apache.flume.Context;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.source.AbstractSource;

public class MySource extends AbstractSource implements Configurable, PollableSource {

    // 定义全局的前缀和后缀
    private String prefix;
    private String suffix;

    /**
     * 1.接收数据（for循环造数据）
     * 2.封装为事件
     * 3.将事件传给channel
     *
     * @return
     * @throws EventDeliveryException
     */
    @Override
    public Status process() throws EventDeliveryException {
        Status status = null;
        try {
            // 1.接收数据
            for (int i = 0; i < 5; i++) {
                // 2.构造事件对象
                SimpleEvent event = new SimpleEvent();
                // 3.给事件设置值
                event.setBody((prefix + "--" + i + "--" + suffix).getBytes());
                // 4.将事件传给channel
                getChannelProcessor().processEvent(event);
                status = Status.READY;
            }
        } catch (Exception e) {
            e.printStackTrace();
            status = Status.BACKOFF;
        }
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return status;
    }

    @Override
    public long getBackOffSleepIncrement() {
        return 0;
    }

    @Override
    public long getMaxBackOffSleepInterval() {
        return 0;
    }

    /**
     * 或取配置文件（XX.conf）中的配置信息
     *
     * @param context
     */
    @Override
    public void configure(Context context) {
        // 读取配置信息给前后缀
        prefix = context.getString("prefix");
        suffix = context.getString("suffix", "neu");
    }
}

1. 创建配置文件 mySource.conf。

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

# Describe/configure the source
a1.sources.r1.type = com.neu.source.MySource
a1.sources.r1.prefix = feiji
a1.sources.r1.suffix = xiaxain

# Describe the sink
a1.sinks.k1.type = logger

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

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

1. 开启监控

bin/flume-ng agent -c conf -f job/mySource.conf -n a1 -Dflume.root.logger=INFO,console

1. 查看结果

自定义 Sink

  使用 Flume 接收数据，并在 Sink 端给每条数据添加前缀和后缀，输出到控制台。前后缀可在 Flume 配置文件中配置。

1. 定义 MySink 类，继承 AbstractSink 类并实现 Configurable 接口。之后将 Maven 项目打成 jar 包上传到 /flume/lib 目录下。

package com.neu.sink;

import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


public class MySink extends AbstractSink implements Configurable {

    // 获取Logger对象
    private Logger logger = LoggerFactory.getLogger(MySink.class);

    // 定义两个属性，前后缀
    private String prefix;
    private String suffix;

    /**
     * 1.获取 Channel
     * 2.从 Channel获取事务和数据
     * 3.发送数据
     *
     * @return
     * @throws EventDeliveryException
     */
    @Override
    public Status process() throws EventDeliveryException {
        // 1.定义返回值
        Status status = null;
        // 2.获取 Channel
        Channel channel = getChannel();
        // 3.从Channel获取事务
        Transaction transaction = channel.getTransaction();
        // 4.开启事务
        transaction.begin();

        try {
            // 5.从Channel获取数据
            Event event = channel.take();
            // 6.处理事件
            if (event != null) {
                String body = new String(event.getBody());
                logger.info(prefix+body+suffix);
            }
            // 7.提交事务
            transaction.commit();
            // 8.成功提交，修改状态信息
            status = Status.READY;
        } catch (ChannelException e) {
            e.printStackTrace();
            // 9.提交事务失败
            transaction.rollback();
            // 10.修改状态
            status = Status.BACKOFF;
        } finally {
            // 11.最终关闭事务
            if (transaction != null) {
                transaction.close();
            }
        }
        // 12.返回状态信息
        return status;
    }

    @Override
    public void configure(Context context) {
        // 读取配置文件，为前后缀赋值
        prefix = context.getString("prefix");
        suffix = context.getString("suffix", "neu");
    }
}

1. 创建配置文件 mySink.conf

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

# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444

# Describe the sink
a1.sinks.k1.type = com.neu.sink.MySink
a1.sinks.k1.prefix = hello--
a1.sinks.k1.suffix = --hello

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

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

1. 开启任务

bin/flume-ng agent -c conf -f job/mySink.conf -n a1 -Dflume.root.logger=INFO,console