flume 几个比较有用的source、sink和decorator

xsmscb

1. Source catalog:
　　1.console[("format")] 控制台输出，可以带输出内容格式的参数



console(“raw”)
console(“json”)
　　
　　2.text("filename"[, format])一次读取文件内容，每一个换行做为一个事件。
　　
　　3.tail("filename"[, startFromEnd=false]{,delim="regex", delimMode="exclude|prev|next"})
　　   读取文件尾部内容跟linux上tail类似。

•    Filename：读取的文件名，
  
•    StartFromEnd 每次重读时，是否重头开始读。默认false每次重头开始重读，
  
•    Delim 是分割参数，
  
•    DelimMode分隔符数据属于哪一部分的指定。
  

　　4、tailDir("dirname"[,fileregex=".*"[, startFromEnd=false[,recurseDepth=0]]]{,delim="regex",delimMode="exclude|prev|next"})
　　对一个目录下面有改动文件的事件发送

• Dirname：目录名
  
• Fileregex：文件名正则表达式匹配，需要符合java中正则表达式规则
  
• StartFromeEnd 跟tail中的参数一样。
  
• RecurseDepth ：指定目录下是否递归对其子目录中文件的监控。指定递归层数。
  
• 其他剩余参数跟tail中一致。
  

　　

2. Sinks catalog

CollectorTier Event Sinks
　　1．   collectorSink("fsdir","fsfileprefix"[, rollmillis[, format]])
　　收集发送过来的事件。

• Fsdir:目录，
  
• Fsfileprefix：文件前缀，
  
• Rollmillis：对于hdfs来就是文件的打开是关闭这段时间
  
• Format是输出文件格式
  

AgentTier Event Sinks
　　1. agentSink[("machine"[,port])]
　　将事件发送到指定机器的指定端口，默认方式是End-to-End,等效于agentE2ESink
　　
　　2. agentE2ESink[("machine"[,port])]
　　先讲event内容写入文件防止在缓冲丢失和点对点确认的数据传输

• Machine: collectorSource的ip，
  
• Port: collectorSource的端口
  

　　
　　3、agentDFOSink[("machine"[,port])]
　　当连接断开时，会把event写到本地，并且会不断重试发送event给对方，把之前未发送成功的event再次发送。
　　
　　4、agentBESink[("machine"[,port])]   
　　BestEffort Agent sink,最有效率的agent sink，顾名思义，就是只管发送，不管有没有发送成功。就可能存在丢失event的情况存在。
　　
　　5.  agentE2EChain("m1[:_p1_]"[, "m2[:_p2_]"[,…]])
　　有多个collector可供选择发送。如果第一丢失连接了，就会向第二发送event，以此类推。并且会不定期回来查看原来没反应的collector是否已经恢复了，如果恢复了，就会跟原来的节点交互了。发送机制跟agentE2ESink一样
　　
　　agentDFOChain和agentBEChain与5类似，另外：autoE2EChain，autoDFOChain，autoBEChain，自动的会想master要可以连接的collector
　　

3. 装饰器catalog

• writeAhead 在sink发送前先写入文件后，在本地缓冲机制，可让接收和发送分别在不同的线程中进行。
  
• ackedWriteAhead[(maxmillis)]
  
• diskFailover[(maxmillis)]：如果event发送失败，就在本地磁盘缓存，然后每隔maxmillis 时间去重试。
  
• ackInjector： 注入ack确认操作。
  
• ackChecker：计算发送组的checksum值，插入到ackInjector中
  
• lazyOpen:只在调用append的时候，才会做真正的open，closed操作
  
• format("pattern")：改变输出格式，具体格式有pattern参数指定。对于需要做checksum的events来说由于改变了数据格式将导致checksum的值也会改变。从而导致消息的丢失。这个装饰器一般用在对消息可靠性要求不是很高，或者reports中。
  
• batch(n,maxlatency)： 缓存n个events 然后统一发送。如果等待时间操作maxlatency，即使未有n个events也会发送。
  
• unbatch：对于上面被batch操作过的events，同步unbatch来拆分开。
  
• gzip：将events事件序列化后压缩
  
• gunzip：对gzip压缩过的events，解压
  
• intervalSampler(n) 每隔n个events 发送一次，算是一个优化手段，特别是通过网络传输时。
  
• probSampler(p)：通过概率p来做决定是否将events接着往下发送
  
• reservoirSampler(k)：对接收到events根据发送时的顺序重新来组织。
  
• delay(ms)  对要发送的events内容延迟ms
  
• choke[(choke-id)]:限制发送速度
  
• 　　value("attr","value"{,escape=true|false})，可以在日志上加一个新的属性，相当于是打tag
  
  
• split("regex",idx,"attr")，可以对日志进行split，取某个字段，然后设置到attr的tag上
  

　　
　　参考自：http://hi.baidu.com/izouying/blog/item/18e6a2d3b0df91c4a8ec9a2b.html