三十九、Flume自定义Source、Sink

谁借莪１个温暖的怀抱￠ 2021-09-07 06:03 364阅读 0赞

上篇文章咱们基于Flume举了几个例子，包括它的扇入扇出等等。这篇文章我们主要来看一下怎样通过自定义Source和Sink来实现Flume的数据采集。关注专栏[《破茧成蝶——大数据篇》][Link 1]，查看更多相关的内容~

--------------------

**目录**

一、自定义Source

1.1 需求说明

1.2 编码实现

1.3 编写Flume配置文件

1.4 测试自定义的Source

二、自定义Sink

2.1 需求说明

2.2 代码实现

2.3 编写Flume的配置文件

2.4 测试自定义Sink

--------------------

# 一、自定义Source #

Source是负责接收数据到Flume Agent的组件，我们不仅可以根据Flume自身提供的API，实现Source的定义，还可以根据项目的实际需求，编写自己的Source，比如Source可以是从网络上下载一个文件，或者是从数据库中查询到的数据。Source组件可以处理各种类型、各种格式的日志数据，包括avro、thrift、exec、jms、spooling directory、netcat、sequence generator、syslog、http、legacy等等。官方提供的source类型已经很多，但是有时候并不能满足实际开发当中的需求，此时我们就需要根据实际需求自定义某些Source。

自定义的Source有两种类型：PollableSource（轮训拉取）与EventDrivenSource （事件驱动）。两者的区别在于PollableSource是通过线程不断去调用process方法，主动拉取消息，而EventDrivenSource是需要触发一个调用机制，即被动等待。在利用PollableSource实现自定义Source时还需要实现Configurable接口，以便在项目中初始化某些配置。下面我们就以实现PollableSource为例进行自定义Source的说明。官方也提供了自定义source的接口，可以点击[这里][Link 2]进行查看~

## 1.1 需求说明 ##

使用Flume接收数据，自定义Source接收数据，并将数据输出到控制台。

## 1.2 编码实现 ##

首先需要导入Maven依赖，如下所示：

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

实现自定义Source的代码如下所示：

package com.xzw.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;
    
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * @author: xzw
     * @create_date: 2021/1/14 15:24
     * @desc: 自定义source
     * @modifier:
     * @modified_date:
     * @desc:
     */
    public class MySource extends AbstractSource implements Configurable, PollableSource {
    
        //定义需要从配置中读取的字段
    
        //两条数据之间的间隔
        private long delay;
        //模拟信息
        private String field;
    
        /**
         * 处理过程
         *
         * @return
         * @throws EventDeliveryException
         */
        public Status process() throws EventDeliveryException {
            try {
                Map<String, String> header = new HashMap<>();
                SimpleEvent event = new SimpleEvent();
                //拿到数据
                for (int i = 0; i < 5; i++) {
                    event.setHeaders(header);
                    event.setBody((field + i).getBytes());
                    getChannelProcessor().processEvent(event);
                    Thread.sleep(delay);
                }
            } catch (Exception e) {
                return Status.BACKOFF;
            }
    
            return Status.READY;
        }
    
        /**
         * 回滚之后睡眠增加多长时间（每回滚一次增加多长时间）
         *
         * @return
         */
        public long getBackOffSleepIncrement() {
            return 0;
        }
    
        /**
         * 最大的失败睡眠间隔
         *
         * @return
         */
        public long getMaxBackOffSleepInterval() {
            return 0;
        }
    
        /**
         * 配置方法
         *
         * @param context
         */
        public void configure(Context context) {
            delay = context.getLong("delay", 2000l);
            field = context.getString("field", "xzw");
        }
    
        public static void main(String[] args) {
    
        }
    }

## 1.3 编写Flume配置文件 ##

自定义flume-mysource.conf配置文件并添加如下内容：

a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    a1.sources.r1.type = com.xzw.source.MySource
    a1.sources.r1.delay = 5000
    #a1.sources.r1.field = xzw
    
    a1.sinks.k1.type = logger
    
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

## 1.4 测试自定义的Source ##

将写好的代码打包上传到Flume的lib目录下，我们这里的路径是/opt/modules/flume/lib。使用如下命令启动Flume进行测试：

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

可以发现在控制天打印出了我们传输的数据：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70][]

# 二、自定义Sink #

Sink组件目的地包括hdfs、logger、avro、thrift、ipc、file、null、HBase、solr等等，但是有时候并不能满足实际开发当中的需求，此时我们就需要根据实际需求自定义某些Sink。官方也提供了自定义source的接口，请点击[这里][Link 3]进行查看~

## 2.1 需求说明 ##

使用Flume接收数据，并在Sink端给每条数据添加前缀和后缀，最后将添加了后缀的数据输出到控制台。

## 2.2 代码实现 ##

实现自定义Sink的代码如下所示：

package com.xzw.source;
    
    import org.apache.flume.*;
    import org.apache.flume.conf.Configurable;
    import org.apache.flume.sink.AbstractSink;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    /**
     * @author: xzw
     * @create_date: 2021/1/15 10:09
     * @desc: 自定义sink
     * @modifier:
     * @modified_date:
     * @desc:
     */
    public class MySink extends AbstractSink implements Configurable {
    
        //创建Logger对象
        private static final Logger LOG = LoggerFactory.getLogger(AbstractSink.class);
    
        //声明前缀和后缀
        private String prefix = "";
        private String suffix = "";
    
        @Override
        public Status process() throws EventDeliveryException {
            //声明返回值状态信息
            Status status = null;
    
            //获取当前sink绑定的channel
            Channel channel = getChannel();
    
            //获取事务
            Transaction transaction = channel.getTransaction();
    
            //开启事务
            transaction.begin();
    
            try {
                //声明事件
                Event take;
                while ((take = channel.take()) == null) {
                    Thread.sleep(200);
                }
    
                //处理事件
                LOG.info(prefix + new String(take.getBody()) + suffix);
    
                //事务提交
                transaction.commit();
                status = Status.READY;
            } catch (Throwable e) {
                //当遇到异常的时候，回滚事务
                transaction.rollback();
                status = Status.BACKOFF;
    
                if (e instanceof Error) {
                    throw (Error) e;
                }
            } finally {
                //关闭事务
                transaction.close();
            }
    
            return status;
        }
    
        @Override
        public void configure(Context context) {
            //读取配置文件内容，有默认值
            prefix = context.getString("prefix", "PREFIX:");
            //读取配置文件内容，没有默认值
            suffix = context.getString("suffix");
    
        }
    
        public static void main(String[] args) {
    
        }
    }

## 2.3 编写Flume的配置文件 ##

编写flume-mysink.conf配置文件，文件内容如下所示：

a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    a1.sources.r1.type = netcat
    a1.sources.r1.bind = localhost
    a1.sources.r1.port = 44444
    
    a1.sinks.k1.type = com.xzw.source.MySink
    a1.sinks.k1.suffix = :SUFFIX
    
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

## 2.4 测试自定义Sink ##

将编写好的代码打包上传到Flume的lib目录下，使用如下命令启动Flume：

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

通过测试可以发现，数据通过自定义的Sink输出到控制台：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 1][]

OK，本文就介绍到这里，自定义的Source、Sink这部分比较简单，你们在这个过程中遇到了什么问题，欢迎留言，让我看看你们遇到了什么问题~

[Link 1]: https://blog.csdn.net/gdkyxy2013/category_10006538.html
[Link 2]: https://flume.apache.org/FlumeDeveloperGuide.html#source
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70]: /images/20210726/3c3a56645f6542309538422eaaa00a26.png
[Link 3]: https://flume.apache.org/FlumeDeveloperGuide.html#sink
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2dka3l4eTIwMTM_size_16_color_FFFFFF_t_70 1]: /images/20210726/1d2deb3c03034f4ca901ffc9a511e76b.png