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上
参考自: