在一次实验过程中,使用flume 1.7采集本地的数据到hdfs文件系统时,由于配置文件不合理,导致出错。错误如下:

[WARN - org.apache.hadoop.hdfs.DFSOutputStream$DataStreamer.closeResponder(DFSOutputStream.java:611)] Caught exception
java.lang.InterruptedException
at java.lang.Object.wait(Native Method)
at java.lang.Thread.join(Thread.java:1281)
at java.lang.Thread.join(Thread.java:1355)
at org.apache.hadoop.hdfs.DFSOutputStream$DataStreamer.closeResponder(DFSOutputStream.java:609)
at org.apache.hadoop.hdfs.DFSOutputStream$DataStreamer.endBlock(DFSOutputStream.java:370)
at org.apache.hadoop.hdfs.DFSOutputStream$DataStreamer.run(DFSOutputStream.java:546)

Flume配置文件如下:

a1.sources=r1
a1.sinks=k1
a1.channels=c1

## describe sources
a1.sources.r1.type=spooldir
a1.sources.r1.spoolDir=/home/hadoop/data/flumeData
a1.sources.r1.fileHeader = false
a1.sources.r1.deserializer.maxLineLength=10240

## describe channel
a1.channels.c1.type=memory

##describe sinks
a1.sinks.k1.type=hdfs
a1.sinks.k1.hdfs.path=/hadoopProject01/%y-%m-%d/
a1.sinks.k1.hdfs.filePrefix = beicai
a1.sinks.k1.hdfs.fileType=DataStream
#每个批次刷新到hdfs的events的数据
a1.sinks.k1.hdfs.batchSize =10
#如果设置为0，则表示不根据events数据来滚动文件
a1.sinks.k1.hdfs.rollCount= 0
#当临时文件达到该大小的时候（单位为bytes）时，滚动成目标文件
a1.sinks.k1.hdfs.rollSize=52428800
#如果设置为0，则表示不根据时间来滚动文件
a1.sinks.k1.hdfs.rollInterval=0
#是否启用时间上的"舍弃"
a1.sinks.k1.hdfs.round = false
#5分钟后滚动为一个文件
a1.sinks.k1.hdfs.roundValue = 5
a1.sinks.k1.hdfs.roundUnit = minute
a1.sinks.k1.hdfs.useLocalTimeStamp=true

#bind
a1.sources.r1.channels =c1
a1.sinks.k1.channel=c1

本次采用的是直接监控目录下文件的变化,如果有新文件上传到此目录就采集。

处理结果:

因为channel是Memory 类型,所以如果文件很大,就必须保证内存足够,否则发生错误很正常。

还有就是重新调整一下,a1.sinks.k1.hdfs.rollSize=52428800
rollSize可以调小一些。