首页 理论教育 如何创建自定义Sink

如何创建自定义Sink

时间:2023-06-26 理论教育 版权反馈
【摘要】:Sink是完全事务性的。官方提供了自定义sink的接口。根据官方说明可知,自定义MySink需要继承AbstractSink类并实现Configurable接口。①将项目打包,上传到flume的lib目录下。②创建并编写flume配置文件mysink.conf。

如何创建自定义Sink

Sink是完全事务性的。在从Channel批量删除数据之前,每个Sink用Channel启动一个事务。批量事件一旦成功写出到存储系统或下一个Flume Agent,Sink就利用Channel提交事务。事务一旦被提交,该Channel从自己的内部缓冲区删除事件。Sink不断地轮询Channel中的事件且批量地移除它们,并将这些事件批量写入到存储或索引系统或者被发送到另一个Flume Agent。Sink组件目的地包括HDFS、logger、avro、thrift、ipc、file、null、HBase、solr、自定义。官方提供的Sink类型已经很多,但是有时候并不能满足实际开发当中的需求,此时我们就需要根据实际需求自定义某些Sink,读取Channel数据写入MySQL或者其他文件系统。

(1)自定义sink的接口

官方提供了自定义sink的接口(https://flume.apache.org/FlumeDeveloperGuide.html#sink)。

根据官方说明可知,自定义MySink需要继承AbstractSink类并实现Configurable接口。其中,实现方法configure()和process()的含义如下:

configure(Context context)//初始化context(读取配置文件内容)

process()//从Channel读取获取数据(event),这个方法将被循环调用。

(2)自定义sink实现。

同样,本节我们仍以案例形式说明自定义sink实现的具体过程。

任务要求:使用flume接收数据,并在Sink端给每条数据添加前缀和后缀,输出到控制台。前后缀可在flume任务配置文件中配置。

自定义sink具体实现步骤如下:

1)创建自定义Sink。

①创建maven工程,添加pom依赖。

②MySink类实现。

2)打包测试。

①将项目打包,上传到flume的lib目录下(具体过程参见7.4.1)。

②创建并编写flume配置文件mysink.conf。

1.sudo touch mysink.conf

2.sudovim mysink.conf

在mysink.conf文件中,添加以下内容:

# Name the components on this agent

a1.sources=r1

a1.sinks=k1(www.xing528.com)

a1.channels=c1

# Describe/configure the source

a1.sources.r1.type=netcat

a1.sources.r1.bind=127.0.0.1

a1.sources.r1.port=4444

# Describe the sink

a1.sinks.k1.type=sink.MySink

# a1.sinks.k1.prefix=jhxxb:

a1.sinks.k1.suffix=:end

# 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

③在终端输入以下shell命令,启动flume运行测试。

1.cd/usr/local/flume/bin

2.bin/flume-ng agent--conf conf/--name a1--conf-file/tmp/flume

-job/sink/mysink.conf-Dflume.root.logger=INFO,console

④向监听端口发送数据。

1.nc 127.0.0.1 4444

2.123

终端输出结果如图7-36所示。

图7-36 终端输出结果

免责声明:以上内容源自网络,版权归原作者所有,如有侵犯您的原创版权请告知,我们将尽快删除相关内容。

我要反馈