《Flume Source组件实战—Avro、Spool、Exec（详细图文）》一文中，我们介绍了集中不同的Source 组件的使用方式，Source监控取到的数据大多数都是通过LoggerSink输出Cli端界面上，并没有将这些数据真正下沉落地，那么这里LoggerSink记录INFO级别的日志，一般多数用来进行系统调试。在本文中，我们将着重介绍在生产环境中常用的sink组件：HDFS Sink

Flume Sink的目的是从Flume Channel中获取数据然后输出到存储或者其他Flume Source中。Flume Agent启动的时候，它会为每一个Sink都启动一个SinkRunner的对象，SinkRunner.start()方法会启动一个新的线程去管理每一个Sink的生命周期。每一个Sink需要实现start()、Stop()和process()方法。本文只是例举几个Sink而已，Flume本身含有丰富的Source 和Sink组件，并且支持随意组合，只要你按照配置的规则来生产数据、传输数据即可，使用非常方便。

HDFS Sink 实战

在本节我们要实现Flume实时监测日志文件，并将新增的日志数据写入到HDFS文件中

读其名可知，HDFS Sink就是Sink将从Channel中get到的时间写入到Hadoop分布式文件系统HDFS，从而完成日志文件的持久化操作。目前Flume支持创建文本文件和序列化文件，这些文件可以按照指定的时间或数据量或事件的数量为基础进行存放，HDFS的目录路径可以包含将要由HDFS替换格式的转移序列用以生成存储事件的目录/文件名。agent从WebServer拿到数据之后再存放到HDFS，整个过程如下图所示：

1、添加配置文件内容cd /usr/flume/conf 在conf目录下，将我们之前写的exec.conf直接copy过来进行修改即可，cp exec.conf exec_hdfs_sink.conf，增加以下内容：

#example.conf: A single-node flume configuration
#use hdfsSink 

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

#配置source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /usr/flume/logs/exec_tail_test.log
a1.sources.r1.channels = c1

#配置HDFS sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.type = hdfs://master:9000/flume/data/logs/
a1.sinks.k1.hdfs.fileType = DataStream
a1.sinks.k1.hdfs.writeFormat = TEXT
#设置输出文件的前缀
a1.sinks.k1.hdfs.filePrefix = flumeHdfs
#设置输出文件的后缀
a1.sinks.k1.hdfs.fileSuffix = .log
a1.sinks.k1.hdfs.batchSize = 1000
a1.sinks.k1.hdfs.rollSize = 10240
a1.sinks.k1.hdfs.rollCount = 0
a1.sinks.k1.hdfs.rollInterval = 1
a1.sinks.k1.hdfs.useLocalTimeStamp = true

#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

参数解释，但这三个参数都不是必须的，如不明确指定，Flume有默认给定值：

• a1.sinks.k1.hdfs.rollSize = 10240 //设置文件大小，当文件达到一定大小才传输（默认为1024字节）
• a1.sinks.k1.hdfs.rollCount = 0 //设置接收的时间数目，当文件写满指定数量之后传输（默认10个事件）
• a1.sinks.k1.hdfs.rollInterval = 1 //设置文件间隔多久发送，时间越长，获取内容越多（默认30s），但是这个值不能设置为0，否则flume在连接hdfs时会报超时连接的error，导致程序反复在重连，cli界面处于等到状态，如下图：

2、启动。使用命令，启动flume agent。注意：在启动之前我们首先要启动hadoop，很多人在使用的时候直接运行了start-all.sh命令启动了yarn和hdfs，其实没必要，我们只需要start-dfs.sh 启动分布式文件系统HDFS即可，还可节约VM的资源占用。

[root@master ~]# flume-ng agent -n a1 -c /usr/flume/conf/ --conf-file /usr/flume/conf/exec_hdfs_sink.conf -Dflume.root.logger=INFO,console 

3、成功启动之后，如下所示，系统程序界面会一直处于监控状态，监控该命令下文件是否出现变动，从上面的日志信息可以看出，相关组件进程均已成功启动。

4、因为这里我们选用的Source组价是Exec Source，监控的命令是tail -F /usr/flume/logs/exec_tail_test.log，在本次实验之前，我们logs文件夹下不含有exec_tail_test.log文件，因此我们首先先创建一个并写入下面图片中的测试内容，vi /usr/flume/logs/exec_tail_test.log

在点击保存文件之后，会发现agent的界面会出现连接hdfs，并创建我们在配置文件中设定好的文件路径、文件名称等。注：在实验中我们发现，配置文件中设定的HDFS目录，并不需要提前创建好然后赋成777权限，如果hdfs中不存在此文件夹，Flume会自行创建。如下图所示。

5、登录master：50070端口，去查看HDFS的目录结构，并确认是否已经写入我们测试的内容：

从上图我们可以看出，不管是文件夹还是写出的文件都已创建好，再查看一下文件的内容：hadoop fs -cat /flume/logs/flumeHdfs.1544687430036.log

可以看出测试内容已经全部写入，HDFS Sink测试成功。

测试时错误：

当向检测文件添加数据时，采用命令echo命令循环生成100+条数据，会报以下错误，提示Channel的Memo已经满了

[root@master logs]# for i in {1..121};do echo "$i new insert data" >> exec_tail_test.log;done  

再去HDFS的文件系统上查看目录，会发现ChannelSize没有减少，hdfs里的数据也一直在重复被写入，但是前台界面报错，肯定不符合生产需要。于是我们开始看配置文件，发现我们的a1.channels.c1.transactionCapacity = 100，发现这个值设置成了100，而我们要同时写入超过100条的数据，因此会产生channel队列满了的错误。

于是查阅资料，把整个问题都解决，发现这个错误关乎三个参数：channels.capacity 、channels.transactionCapacity 、sinks.hdfs.batchSize ；事情的真相原来是这样，这个Channels的capacity是指整个队列的缓存最大容量，transactionCapacity则是指事务event的最大容量，即每次传输的event最大为多少，而这个sink的batchsize是什么意思呢，就是sink会一次从channel中取多少个event去发送，而这个发送是要最终以事务的形式去发送的，因此这个batchsize的event会传送到一个事务的缓存队列中（takeList），这是一个双向队列，这个队列可以在事务失败时进行回滚（也就是把取出来的数据吐memeryChannel的queue中），它的初始大小就是transactionCapacity定义的大小，源码中有：takeList = new LinkedBlockingDeque<Event>(transCapacity); 上面的错误的原因就是，本来队列的最大容量只有100，而我想一次去121个events塞到takeList中。从上述分析中可知，三个参数容量大小应该是：capacity>transactionCapacity>batchSize ;因此，要是想同时将测试用例的121条数据成功写入到hdfs的话，应该将transactionCapacity的值设置的稍微大一些，后来本小厨改成了500。

flume产生大量小文件的解决方法

从上面的HDFS的目录结构可以看出，HDFS的block块大小为128M，但是每次不满1k就输出到第二个文件中，我们可以用两种方法控制：第一控制输出文件的大小a1.sinks.k1.hdfs.rollSize = 10240，这里设置的是10M，但是每到10M就输出了，是因为我们下面设置的时间为1s输出一次；第二个就是通过控制输出时间，a1.sinks.k1.hdfs.rollInterval = 1，可以将时间延长一些。

总结

在使用Source Channel Sink组合时，要特别注意各个参数的配合使用。本例中，我们使用的是Exec Source ，要是使用spool Source基本流程也是一样，只需更改配置文件中的Source组件定义，如下：

a1.sources.r1.channels = c1
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /usr/flume/logs
a1.sources.r1.fileHeader = false