分享

flume1.7如何实现自定义sink


问题导读

1.实现自定义sink需要继承哪个类?
2.本文实现了什么功能?
3.开发完成后,如何配置使用?






用户不仅可以自定义flume的source,还可以自定义flume的sink,用户自定义sink在flume中只需要继承一个基类:AbstractSink,然后实现其中的方法就可以了,例如我现在的需求是只要用户使用我自定义的sink,那么它需要提供一个文件名称,该名称如果有具体路径的话,需要填写路径的全名称,实现的功能就是将数据保存到该文件名称中,因为用户是基于配置的,所以无须理会source的来源问题,下面我们来编写代码:

1、加入工程所需jar包:flume-ng-configuration-1.7.0.jar、flume-ng-core-1.7.0.jar、flume-ng-sdk-1.7.0.jar

2、编写我们自定义的类,实现我们的需求:

[mw_shl_code=java,true]
package com.harderxin.flume.test;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;

import org.apache.flume.Channel;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Transaction;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MySinks extends AbstractSink implements Configurable {
    private static final Logger logger = LoggerFactory.getLogger(MySinks.class);
    private static final String PROP_KEY_ROOTPATH = "fileName";
    private String fileName;

    @Override
    public void configure(Context context) {
        fileName = context.getString(PROP_KEY_ROOTPATH);
    }

    @Override
    public Status process() throws EventDeliveryException {
        Channel ch = getChannel();
        //get the transaction
        Transaction txn = ch.getTransaction();
        Event event =null;
        //begin the transaction
        txn.begin();
        while(true){
            event = ch.take();
            if (event!=null) {
                break;
            }
        }
        try {

            logger.debug("Get event.");

            String body = new String(event.getBody());
            System.out.println("event.getBody()-----" + body);
            
            String res = body + ":" + System.currentTimeMillis() + "\r\n";
            File file = new File(fileName);
            FileOutputStream fos = null;
            try {
                fos = new FileOutputStream(file, true);
            } catch (FileNotFoundException e) {
                e.printStackTrace();
            }
            try {
                fos.write(res.getBytes());
            } catch (IOException e) {
                e.printStackTrace();
            }
            try {
                fos.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
            txn.commit();
            return Status.READY;
        } catch (Throwable th) {
            txn.rollback();

            if (th instanceof Error) {
                throw (Error) th;
            } else {
                throw new EventDeliveryException(th);
            }
        } finally {
            txn.close();
        }
    }
}

[/mw_shl_code]
3、将工程打成jar包,放在flume解压后的lib文件中,依赖包不需要放进去,因为flume中的lib目录已经存在了
4、编写我们的配置文件:

[mw_shl_code=bash,true]# 指定Agent的组件名称  
a1.sources = r1  
a1.sinks = k1  
a1.channels = c1  
  
# 指定Flume source(要监听的路径)  
a1.sources.r1.type = netcat  
a1.sources.r1.bind = localhost  
a1.sources.r1.port = 5678  
  
# 指定Flume sink  
a1.sinks.k1.type = com.harderxin.flume.test.MySinks  
a1.sinks.k1.fileName = D://flume-test//sink//mysinks.txt  
  
# 指定Flume channel  
a1.channels.c1.type = memory  
a1.channels.c1.capacity = 1000  
a1.channels.c1.transactionCapacity = 100  
  
# 绑定source和sink到channel上  
a1.sources.r1.channels = c1  
a1.sinks.k1.channel = c1  [/mw_shl_code]

说明:配置文件中的source是用户在控制台telnet命令监听本地的5678端口,然后根据输入的信息,该信息根据我们自定义的sink保存到我们配置的文件中,文件中定义的a1.sinks.k1.fileName属性,就是我们自定义sink的属性,该属性可以让用户自己配置,对应的目录要提前创建好
5、根据命令,启动flume,启动命令在前面的文章中有提到过,然后进行测试,打开cmd命令,使用telnet 监听5678端口,然后在里面输入信息后回车,在自定义的mysinks.txt文件中就会有我们的信息输出来了:
20170712171052534.png

自定义sinks类详解:
1、用户自定义的sink实现Configurable接口,实际上是实现里面的configure(Context context)方法,主要是获取用户配置的一些信息,如果我们还有很多的属性需要用户自己设置,那么我们可以在这个方法中将用户定义的参数取出来,context类中提供了很多get方法,例如getString、getLong、getBoolean等
2、核心的处理逻辑是在process方法中,getChannel方法在父类AbstractSink中已经实现,相当于取得输送信息到sink的Channel对象,然后它里面提供事务操作方法:getTransaction()和取出消息Event的方法:take(),这两个方法在其中很重要,取得事物对象可以保证该信息被自定义的sink成功消费,成功消费后,使用commit方法提交事务,那么Event将从channel队列中删除掉,如果没有成功消费,那么使用rollback方法进行回滚,该Event将还会保留在Channel的队列中,以便下次消费,保证消息不会出现遗漏现象
take方法主要是取出消息Event,在flume中也可以叫做事件,然后通过getBody()方法,获得消息的详细内容,就可以进行我们的功能实现了,保存到文件或者插入到数据库等等
3、对比自定义source和自定义sink的process方法:
自定义Source:通过getChannelProcessor方法获得ChannelProcessor对象,然后通过processEvent方法将消息转换为flume的Event对象传递给Channel处理
自定义sink:通过getChannel方法获得Channel对象,然后通过take方法从Channel中取出Event,然后转换为我们需要的消息数据进行处理
source的process方法相当于Event的生产者,不断向Channel中发送Event,而sink的process方法相当于Event的消费者,不断从Channel中取出Event进行处理
当然,我们可以在配置中使用自定义source和自定义sink相结合的方式,根据需求实现我们的功能,配置如下:

[mw_shl_code=bash,true]
# 指定Agent的组件名称
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# 指定Flume source(要监听的路径)
a1.sources.r1.type = com.harderxin.flume.test.MySource

# 指定Flume sink
a1.sinks.k1.type = com.harderxin.flume.test.MySinks
a1.sinks.k1.fileName = D://flume-test//sink//winlog.txt

# 指定Flume channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
a1.channels.c1.byteCapacityBufferPercentage = 20
a1.channels.c1.byteCapacity = 800000

# 绑定source和sink到channel上
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1[/mw_shl_code]
   到此,是不是感觉Flume很强大,Flume号称是apache的顶级项目,还有很多值得研究学习的地方,我还在这条道路上不断前进着,继续加油..........




欢迎加入about云群425860289432264021 ,云计算爱好者群,关注about云腾讯认证空间

已有(1)人评论

跳转到指定楼层
wangxiaojian 发表于 2018-1-29 08:41:15
  学到了很多 赞!
回复

使用道具 举报

您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关闭

推荐上一条 /2 下一条