Flume参考文档汇总：配置文件、command shell、Source、Sinks配置项、属性说明

问题导读：
agent日志路径如何配置？
flume.collector.port的作用是什么？
flume如何异步执行命令？
collectorSink( "fsdir","fsfileprefix",rollmillis)含义是什么？

Flume配置文件(flume-site.conf)

1、 watchdog

watchdog.restarts.max

watchdog每分钟重启的最大数???

2、 common node

flume.config.heartbeat.period	node发送心跳周期,默认5000(毫秒)
flume.node.status.port	node web端口
flume.node.heartbeat.backoff.ceiling	node向master发送心跳之后等待反馈的最大时长,默认60000(毫秒)
flume.node.http.autofindport	如果已有node启动，允许第二个node自动选择一个未使用的端口做web服务。多个node的界面端口从35862、35863向后延续

3、agent

flume.agent.logdir	agent日志路径
flume.agent.logdir.maxage	当前处于打开状态agent日志文件收集信息的时长,在这之后该日志文件将会被关闭，并将数据发送到网络，默认10000(毫秒)
flume.agent.logdir.retransmit	在end-to-end模式下agent向collector发送数据失败后再次发送的间隔时长，默认60000(毫秒)，建议至少是flume.collector.roll.millis的两倍
flume.agent.failover.backoff.initial	当primary sink(可以认为是第一collector)故障后，重启primary sink的一个延迟时间，在此期间，agent将把数据发送到secondary sink(可能是第二collector)
flume.agent.failover.backoff.max	在一定时限内尝试链接故障节点失败后，agent将把数据转发向备用节点

4、collector

flume.collector.event.host	默认collector地址
flume.collector.port	默认collector端口
flume.collector.dfs.dir	最终数据发向目录(默认)，可以是本地，可以是hdfs，默认是/tmp
flume.collector.dfs.compress.codec	压缩格式GzipCodec, DefaultCodec (deflate), BZip2Codec，默认是None
flume.collector.roll.millis	hdfs文件切换(关闭后新建)的时长
flume.collector.output.format	collector发送数据格式avro, avrojson(默认), avrodata…

5、master

flume.master.servers	用逗号分隔多个master地址列表
flume.master.store	master配置存储方式(zookeeper/memory) zookeeper保证master的配置在多master节点之间同步，memory则保存在内存中，其配置随着master宕机而丢失
flume.master.serverid	master的唯一标识
flume.master.http.port	http端口
flume.master.heartbeat.missed.max	判断节点失效的最大未达心跳数
flume.master.savefile	当前flume配置文件的路径，默认conf/current.flume
flume.master.savefile.autoload	启动时是否加载current.flume，默认false
flume.master.gossip.period	master通信周期(毫秒)
flume.master.heartbeat.rpc	THRIFT/AVRO
flume.event.rpc	THRIFT/AVRO
flume.report.server.rpc.type	THRIFT/AVRO

6、zookeeper

flume.master.zk.logdir

zookeeper日志路径

7、thrift

flume.thrift.socket.timeout.ms

thrift网络连接超时时间(毫秒)

command shell(flume command)

help	帮助
connect master:port	登录master
config logicalnode source sink	为逻辑节点配置一个source到sink的映射
getnodestatus	获得节点状态(HELLO, CONFIGURING, ACTIVE, IDLE, ERROR, DECOMMISSIONED, LOST ) HELLO, node启动时 CONFIGURING, node被配置后 ACTIVE, 一个event从source送达到sink IDLE, source中所有evnet发送完毕后 ERROR, 节点故障退出，数据没有flush DECOMMISSIONED, node被master移除 LOST, master长时间未收到node心跳
getconfigs	获得配置
getmappings [physical node]	如果physical node参数被省略，将显示所有logical node到physical node的映射关系
exec	同步执行命令
Source file	执行脚本.
submit	异步执行命令
wait ms [cmdid]	设定一个时间，周期检查命令进程的状态(success or failure)
waitForNodesActive ms node1 [node2 […]]	设定一个时间，检查node是否处于使用(configuring, active)状态
waitForNodesDone ms node1 [node2 […]]	设定一个时间，检查node是否处于未用(IDLE, ERROR, LOST)状态
quit	退出

command shell(exec & submit command)

双引号	包含转义字符的java string
单引号	能引住除单引号之外的所有字符
noop	touch master, 不做操作
config logicalnode source sink	为逻辑节点配置source到sink的映射
multiconfig flumespec
unconfig logicalnode	取消逻辑节点的配置，影响master调整故障转移列表(failover list)
refreshAll logicalnode	刷新
save filename	保存current configuration到master硬盘
load filename	从master中加载current configuration
map physicalnode logicalnode	配置物理节点到逻辑节点的映射关系，master的配置将被同步到logicalnode
spawn physicalnode logicalnode	恢复
decommission logicalnode
unmap physicalnode logicalnode	取消映射
unmapAll	全部取消
purge logicalnode	清除状态，类似重启一个logical node, 适用于(DECOMMISSIONED、 LOST)状态
purgeAll	清除所有logical node的状态

Flume Source

1、Flume’s Tiered Event Sources

collectorSource[(port)]	Collector source，监听端口汇聚数据
autoCollectorSource	通过master协调物理节点自动汇聚数据
logicalSource	逻辑source，由master分配端口并监听rpcSink

2、Flume’s Basic Sources

null
console	监听用户编辑历史和快捷键输入，只在node_nowatch模式下可用
stdin	监听标准输入，只在node_nowatch模式下可用，每行将作为一个event source
rpcSource(port)	由rpc框架(thrift/avro)监听tcp端口
text("filename")	一次性读取一个文本，每行为一个event
tail("filename"[, startFromEnd=false])	每行为一个event。监听文件尾部的追加行，如果startFromEnd为true，tail将从文件尾读取，如果为false，tail将从文件开始读取全部数据
multitail("filename"[, file2 [,file3… ] ])	同上，同时监听多个文件的末尾
tailDir("dirname"[, fileregex=".*"[, startFromEnd=false[, recurseDepth=0]]])	监听目录中的文件末尾，使用正则去选定需要监听的文件（不包含目录），recurseDepth为递归监听其下子目录的深度
seqfile("filename")	监听hdfs的sequencefile，全路径
syslogUdp(port)	监听Udp端口
syslogTcp(port)	监听Tcp端口
syslogTcp1(port)	只监听Tcp端口的一个链接
execPeriodic("cmdline", ms)	周期执行指令，监听指令的输出，整个输出都被作为一个event
execStream("cmdline")	执行指令，监听指令的输出，输出的每一行被作为一个event
exec("cmdline"[,aggregate=false[,restart=false[,period=0]]])	执行指令，监听指令的输出，aggregate如果为true，整个输出作为一个event如果为false，则每行作为一个event。如果restart为true，则按period为周期重新运行
synth(msgCount,msgSize)	随即产生字符串event,msgCount为产生数量，msgSize为串长度
synthrndsize(msgCount,minSize,maxSize)	同上，minSize – maxSize
nonlsynth(msgCount,msgSize)
asciisynth(msgCount,msgSize)	Ascii码字符
twitter("username","pw"[,"url"])	尼玛twitter的插件啊
irc("server",port, "nick","chan")
scribe[(+port)]	Scribe插件
report[(periodMillis)]	生成所有physical node报告为事件源

Flume Sinks

1、Flume’s Collector Tier Event Sinks

collectorSink( "fsdir","fsfileprefix",rollmillis)

collectorSink，数据通过collector汇聚之后发送到hdfs, fsdir 是hdfs目录，fsfileprefix为文件前缀码

2、Flume’s Agent Tier Event Sinks

agentSink[("machine"[,port])]	Defaults to agentE2ESink，如果省略，machine参数，默认使用flume.collector.event.host与flume.collector.event.port作为默认collecotr（以下同此）
agentE2ESink[("machine"[,port])]	执着的agent，如果agent发送event没有收到collector成功写入的状态码，该event将被agent重复发送，直到接到成功写入的状态码
agentDFOSink[("machine" [,port])]	本地热备agent，agent发现collector节点故障后，不断检查collector的存活状态以便重新发送event，在此间产生的数据将缓存到本地磁盘中
agentBESink[("machine"[,port])]	不负责的agent，如果collector故障，将不做任何处理，它发送的数据也将被直接丢弃
agentE2EChain("m1[:_p1_]" [,"m2[:_p2_]"[,…]])	指定多个collector提高可用性。当向主collector发送event失效后，转向第二个collector发送，当所有的collector失败后，它会非常执着的再来一遍...
agentDFOChain("m1[:_p1_]"[, "m2[:_p2_]"[,…]])	同上，当向所有的collector发送事件失效后，他会将event缓存到本地磁盘，并检查collector状态，尝试重新发送
agentBEChain("m1[:_p1_]"[, "m2[:_p2_]"[,…]])	同上，当向所有的collector发送事件失效后，他会将event丢弃
autoE2EChain	无需指定collector, 由master协调管理event的流向
autoDFOChain	同上
autoBEChain	同上

3、Flume’s Logical Sinks

logicalSink("logicalnode")

4、Flume’s Basic Sinks

在不使用collector收集event的情况下，可将source直接发向basic sinks

null	null
console[("formatter")]	转发到控制台
text("txtfile" [,"formatter"])	转发到文本文件
seqfile("filename")	转发到seqfile
dfs("hdfspath")	转发到hdfs
customdfs("hdfspath"[, "format"])	自定义格式dfs
+escapedCustomDfs("hdfspath", "file", "format")
rpcSink("host"[, port])	Rpc框架
syslogTcp("host"[,port])	发向网络地址
irc("host",port, "nick", "chan")