Flume调优

这是一个关于池子的故事。有一个池子，它一头进水，另一头出水，进水口可以配置各种管子，出水口也可以配置各种管子，可以有多个进水口、多个出水口。水术语称为Event，进水口术语称为Source、出水口术语成为Sink、池子术语成为Channel，Source+Channel+Sink，术语称为Agent。如果有需要，还可以把多个Agent连起来。

 

channel 

channel 是flume的中间数据缓存管道，有点类似kafka的机制，因此个组件的性能很重要。

我在项目中主要采用的是menmory channel,原因是数据量大，要求极大的数据吞吐量和速度，但是有一点不好的是

如果一旦flume进程down掉，是没有“续点传输”的机制的，filechannel 和它正好相反。 

 

关键参数讲解：

1.  (1)   capacity  :   存储在channel中的events的最大数量
2.  (2)   transactionCapacity ： 每次数据由channel到sink传输的最大events的数量
3.  (3)   byteCapacity  ：该channel的内存大小，单位是 byte 。

"PollableSourceRunner-SQLSource-sourceOne" java.lang.OutOfMemoryError:

 

       通过排查问题后得出造成此类问题的原因是，项目的一个模块在批量导入数据，程序需要New一个很大的对象，GC发现内存不够后，对内存进行回收，但是发现回收后的内存还不能满足大对象的需要。应此造成了内存溢出。

flume-env.sh

1. export JAVA_OPTS="-Xms100m -Xmx2000m -Dcom.sun.management.jmxremote"

 

Caused by: org.apache.flume.ChannelException: Cannot commit transaction. Byte capacity allocated to store event body 640000.0reached. Please increase heap space/byte capacity allocated to the channel as the sinks may not be keeping up with the sources

 

1. 2017-07-19 10:56:04,366 (PollableSourceRunner-SQLSource-sourceOne) [ERROR - org.apache.flume.source.PollableSourceRunner$PollingRunner.run(PollableSourceRunner.java:156)] Unhandled exception, logging and sleeping for 5000ms
2. org.apache.flume.ChannelException: Unable to put batch on required channel: org.apache.flume.channel.MemoryChannel{name: channelOne}
3. at org.apache.flume.channel.ChannelProcessor.processEventBatch(ChannelProcessor.java:200)
4. at org.keedio.flume.source.SQLSource$ChannelWriter.flush(SQLSource.java:168)
5. at java.io.PrintWriter.flush(PrintWriter.java:320)
6. at com.opencsv.CSVWriter.flush(CSVWriter.java:320)
7. at org.keedio.flume.source.SQLSource.process(SQLSource.java:94)
8. at org.apache.flume.source.PollableSourceRunner$PollingRunner.run(PollableSourceRunner.java:139)
9. at java.lang.Thread.run(Thread.java:745)
10. Caused by: org.apache.flume.ChannelException: Cannot commit transaction. Byte capacity allocated to store event body 640000.0reached. Please increase heap space/byte capacity allocated to the channel as the sinks may not be keeping up with the sources
11. at org.apache.flume.channel.MemoryChannel$MemoryTransaction.doCommit(MemoryChannel.java:123)
12. at org.apache.flume.channel.BasicTransactionSemantics.commit(BasicTransactionSemantics.java:151)
13. at org.apache.flume.channel.ChannelProcessor.processEventBatch(ChannelProcessor.java:192)
14. ... 6 more

 在sink中，channel的transactionCapacity参数不能小于sink的batchsize。