试用flume-ng 1.1

Flume NG 1.x 是Flume 0.9.x的重构版本，基本面目全非了，Master和zookeeper没有了，collector没有了，Web console没有了，只有

• source （avro：很简单使用；exec：使用shell命令）
• sink （我用的hdfs）
• channl

这3个组件，俨然从一个分布式系统变成了传输工具。新的架构如下：

 

 

下面是一个例子（参数经过优化），使用avro作为source，hdfs作为sink，memory作为channel

# Define a memory channel called ch1 on agent1
agent1.channels.ch1.type = memory
agent1.channels.ch1.capacity = 100000
agent1.channels.ch1.transactionCapacity = 100000
agent1.channels.ch1.keep-alive = 30

# Define an Avro source called avro-source1 on agent1 and tell it
# to bind to 0.0.0.0:41414. Connect it to channel ch1.
agent1.sources.avro-source1.channels = ch1
agent1.sources.avro-source1.type = avro
agent1.sources.avro-source1.bind = 0.0.0.0
agent1.sources.avro-source1.port = 41414
agent1.sources.avro-source1.threads = 5

# Define a logger sink that simply logs all events it receives
# and connect it to the other end of the same channel.
agent1.sinks.log-sink1.channel = ch1
agent1.sinks.log-sink1.type = hdfs
agent1.sinks.log-sink1.hdfs.path = hdfs://CNC-XX-R-541:9000/flume/
agent1.sinks.log-sink1.hdfs.writeFormat = Text
agent1.sinks.log-sink1.hdfs.fileType = DataStream
agent1.sinks.log-sink1.hdfs.rollInterval = 0
agent1.sinks.log-sink1.hdfs.rollSize = 60554432
agent1.sinks.log-sink1.hdfs.rollCount = 0
agent1.sinks.log-sink1.hdfs.batchSize = 1000
agent1.sinks.log-sink1.hdfs.txnEventMax = 1000
agent1.sinks.log-sink1.hdfs.callTimeout = 60000
agent1.sinks.log-sink1.hdfs.appendTimeout = 60000

# Finally, now that we've defined all of our components, tell
# agent1 which ones we want to activate.
agent1.channels = ch1
agent1.sources = avro-source1
agent1.sinks = log-sink1

 

启动avro agent，参数“ageng1”在上述配置文件定义。

flume-ng agent --conf ./conf/ -f conf/flume.conf -n agent1

 

client端上传文件

flume-ng avro-client --conf conf -H localhost -p 41414 -F /data/xx.txt

 

心得：

• 感觉就是一个传输工具，使用配置很简单，但是参数还得调，要不报错。
• 上传文件可以自动按照大小、行或者处理时间分割为多个文件。
• 测试了单机上传700MB文件，上传速度和hadoop fs -put速度相同，记录没有丢失。

纠结：

• 输出文件只能在flume.conf定义吗？
• 如何将文件按照规则汇总到HDFS（比如按照客户合并输出）？
• 基于事务的传输保证每个事务内Event（日志行）有保证传到HDFS，但是如果传输整个文件中途中断，还是会有部分数据上传遗留到HDFS。

结论：FlumeNG1.1，尚不能满足复杂业务要求，尽管支持自定义source和sink等组件，系统使用过程中觉得软件不够坚固，非常简单的示例，都报错，让人心里没底，还好源码够简单，这样看来只适合做简单传输。

错误记录：

org.apache.flume.ChannelException: Space for commit to queue couldn't be acquired Sinks are likely not keeping up with sources, or the buffer size is too tight

解决：设置agent1.channels.<channel_name>.keep-alive = 30

资料：

FlumeNG 架构

https://blogs.apache.org/flume/entry/flume_ng_architecture

Flume User Guide

https://people.apache.org/~mpercy/flume/flume-1.2.0-incubating-SNAPSHOT/docs/FlumeUserGuide.html#data-ingestion

--heipark

评论

8 楼 luyee2010 2013-12-04  

Could not find or load main class org.apache.flume.tools.GetJavaProperty

7 楼 java.jianan 2013-03-27  

不好意思，说错了，是 3 M/S 写入dfs，但是和 put 比起来，速度还是很慢，我把参数贴出来：
a1.sinks.k1.hdfs.rollInterval=600
a1.sinks.k1.hdfs.threadsPoolSize=10000
a1.sinks.k1.hdfs.rollSize=21474836480
a1.sinks.k1.hdfs.rollCount=0
a1.sinks.k1.hdfs.callTimeout=80000
a1.sinks.k1.hdfs.batchSize=200000

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 20000000
a1.channels.c1.transactionCapacity=20000000
a1.channels.c1.keep-alive=60

JAVA_OPTS="-Xms61440m -Xmx61440m -Xmn20000m -Xss200k -XX:+UseParallelGC -XX:ParallelGCThreads=12 -XX:+UseParallelOldGC"

6 楼 heipark 2013-03-25  

java.jianan 写道

我测试本地上传1G txt 文件，平均每分钟写入 hdfs 只有3M，有什么办法能提高写的速度呢?

调整sink hdfs参数试试吧，你这个速度不正常，怎么也要每秒10MB数据

5 楼 java.jianan 2013-03-24  

我测试本地上传1G txt 文件，平均每分钟写入 hdfs 只有3M，有什么办法能提高写的速度呢?

4 楼 shenbai 2012-10-11  

avro rpc client 直接调用apendBatch函数，发数据给source，也没有返回机制（同步），怎么控制client发送速度，发的太快，flume处理不过来就会死掉，文件一大就oom，不知道怎么解决呢？

3 楼 long904 2012-08-28  

heipark 写道

long904 写道

感觉这个上传速度很慢，与put根本不是一个等级的。是否哪里配置错误了呢，请教！

兄弟flume-avro-agent使用的是netty编写的，网速速度应该是有保证的。

你说的put速度指什么？

hadoop fs -put

2 楼 heipark 2012-08-27  

long904 写道

感觉这个上传速度很慢，与put根本不是一个等级的。是否哪里配置错误了呢，请教！

兄弟flume-avro-agent使用的是netty编写的，网速速度应该是有保证的。

你说的put速度指什么？

1 楼 long904 2012-08-27  

感觉这个上传速度很慢，与put根本不是一个等级的。是否哪里配置错误了呢，请教！