本文介绍初次使用Flume传输数据到MongoDB的过程,内容涉及环境部署和注意事项。 1 环境搭建 需要jdk、flume-ng、mongodb java driver、flume-ng-mongodb-sink 2 简单原理介绍 这是一个关于池子的故事。有一个池子,它一头进水,另一头出水,进水口可以配置各种管子,出水口也可以配置各种管子,可以有多个进水口、多个出水口。水术语称为Event,进水口术语称为Source、出水口术语成为Sink、池子术语成为Channel,Source+Channel+Sink,术语称为Agent。如果有需要,还可以把多个Agent连起来。 3 Flume配置 (1) env配置 将mongo-java-driver和flume-ng-mongodb-sink两个jar包放到flume\lib目录下,并将路径加入到flume-env.sh文件的FLUME_CLASSPATH变量中; (2) log配置 在调试时,将日志设置为debug并打到文件:flume.root.logger=DEBUG,LOGFILE my_agent.sources.my_source_1.channels = my_channel_1 my_agent.sources.my_source_1.type = exec my_agent.sources.my_source_1.command = python xxx.py my_agent.sources.my_source_1.shell = /bin/bash -c my_agent.sources.my_source_1.restartThrottle = 10000 my_agent.sources.my_source_1.restart = true my_agent.sources.my_source_1.logStdErr = true my_agent.sources.my_source_1.batchSize = 1000 my_agent.sources.my_source_1.interceptors = i1 i2 i3 my_agent.sources.my_source_1.interceptors.i1.type = static my_agent.sources.my_source_1.interceptors.i1.key = db my_agent.sources.my_source_1.interceptors.i1.value = cswuyg_test my_agent.sources.my_source_1.interceptors.i2.type = static my_agent.sources.my_source_1.interceptors.i2.key = collection my_agent.sources.my_source_1.interceptors.i2.value = cswuyg_test my_agent.sources.my_source_1.interceptors.i3.type = static my_agent.sources.my_source_1.interceptors.i3.key = op my_agent.sources.my_source_1.interceptors.i3.value = upsert 字段说明: my_agent.channels.my_channel_1.type = file
my_agent.channels.my_channel_1.checkpointDir = /home/work/flume/file-channel/my_channel_1/checkPoint
my_agent.channels.my_channel_1.useDualCheckpoints = true
my_agent.channels.my_channel_1.backupCheckpointDir = /home/work/flume/file-channel/my_channel_1/checkPoint2
my_agent.channels.my_channel_1.dataDirs = /home/work/flume/file-channel/my_channel_1/data
my_agent.channels.my_channel_1.transactionCapacity = 10000
my_agent.channels.my_channel_1.checkpointInterval = 30000
my_agent.channels.my_channel_1.maxFileSize = 4292870142
my_agent.channels.my_channel_1.minimumRequiredSpace = 524288000
my_agent.channels.my_channel_1.capacity = 100000
字段说明: 要注意的参数是capacity,它指定了池子里可以存放的Event数量,需要根据日志量设置一个合适的值,如果你也采用file-channel,而且磁盘充足,那可以尽可能的设置得大些。 sink配置举例: my_agent.sinks.my_mongo_1.type = org.riderzen.flume.sink.MongoSink my_agent.sinks.my_mongo_1.host = xxxhost my_agent.sinks.my_mongo_1.port = yyyport my_agent.sinks.my_mongo_1.model = dynamic my_agent.sinks.my_mongo_1.batch = 10 my_agent.sinks.my_mongo_1.channel = my_channel_1 my_agent.sinks.my_mongo_1.timestampField = _S 字段说明: model选择dynamic,表示mongodb的db、collection名字采用Event Header中指定的名字。timestampField 字段用于将json串中指定键的值转换为datetime格式存进mongodb,flume-ng-mongodb-sink不支持嵌套key指定(如:_S.y),但可以自己通过修改sink的代码来实现。 my_agent.channels = my_channel_1 my_agent.sources = my_source_1 my_agent.sinks = my_mongo_1 (4) 启动 可以写一个control.sh 脚本来控制flume的启动、关闭、重启。
4 其它 1、监控:官方推荐的监控是ganglia:http:///projects/ganglia/,有图像界面。 2、版本变更:flume 从1.X开始已经不再使用ZooKeeper,在数据可靠性上,提供了E2E(end-to-end)的支持,去掉了重构之前的DFO(store on failure)、BE(best effort)。E2E指的是:在删除channel中的event时,保证event已经传递到了下一个agent或者终点,不过,这里没有提到数据在进入到channel之前如何保证不丢失,像Exec Source这种数据导入channel的方式,需要使用者自己保证。 3、关闭插件:使用Exec Source时,flume重启不会关闭掉旧插件进程,需要自己关闭。 4、Exec Source不能保证数据不丢失,因为这种方式只是把水灌到池子里,不管池子是什么状况, 参见https://flume./FlumeUserGuide.html#exec-source 的 Warning 部分。但是,Spooling directory source 也不一定是个好方法,监控目录,但是注意不能修改文件的名字,不能出现同名覆盖文件,不要出现只有一半内容的文件。传输完成之后,文件会被重命名为xx.COMPLETED,需要有定时清理脚本把这些文件清理掉。重启会导致出现重复event,因为那些被传输到一半的文件没有被设置为完成状态。 5、传输瓶颈:使用flume+mongodb来安全传输大量数据(每秒万条级别的日志不算大数据量,每天几百G的也不算),瓶颈会出现在MongoDB上,特别是Update类型的数据传输。 6、需要修改当前的flume-ng-mongodb-sink 插件:(1)让update支持 $setOnInsert;(2)解决update的 $set、$inc为空时,引发exception的bug;(3)解决批量插入时,因其中一条日志有duplicate exception而导致同批插入的后续日志全部被丢弃的bug。 7、flume跟fluentd很类似,但来自hadoop生态的flume更热门,所以我选择flume。 8、批量部署:先把jdk、flume打包成tar,然后借助python 的 paramiko库,将tar包发到各台机器上,解压、运行。
本文所在:http://www.cnblogs.com/cswuyg/p/4498804.html 1、http://flume./FlumeDeveloperGuide.html 2、《Apache Flume: Distributed Log Collection for Hadoop》
|
|
来自: 陈永正的图书馆 > 《Flume-ng》