攔截器 · JAVA

[TOC] # 介紹攔截器是簡單的插件式組件，設置在source和channel之間。source接收到的事件event，在寫入channel之前，攔截器都可以進行轉換或者刪除這些事件。每個攔截器只處理同一個source接收到的事件。可以自定義攔截器 # flume內置的攔截器 ## 時間戳攔截器 flume中一個最經常使用的攔截器，該攔截器的作用是將時間戳插入到flume的事件報頭中。如果不使用任何攔截器，flume接受到的只有message。時間戳攔截器的配置： ![](https://box.kancloud.cn/bd7dc87161ac8e6db59b6bcb0e4c7f0a_700x343.png) source連接到時間戳攔截器的配置： ~~~ a1.sources.r1.interceptors=i1 a1.sources.r1.interceptors.i1.type=timestamp a1.sources.r1.interceptors.i1.preserveExisting=false ~~~ 事件打印出來 ~~~ 18/07/13 01:08:44 INFO sink.LoggerSink: Event: { headers:{timestamp=1531415324910} body: 68 65 6C 6C 6F 0D hello. } ~~~ 可以看到headers中有添加時間戳一般情況headers里面是空的 ## 主機攔截器主機攔截器插入服務器的ip地址或者主機名，agent將這些內容插入到事件的報頭中。事件報頭中的key使用hostHeader配置，默認是host。主機攔截器的配置： ![](https://box.kancloud.cn/1ca1a6226c9c1bea011b001bbdf3592f_698x411.png) source連接到主機攔截器的配置： ~~~ a1.sources.r1.interceptors=i2 a1.sources.r1.interceptors.i2.type=host a1.sources.r1.interceptors.i2.useIP=false a1.sources.r1.interceptors.i2.preserveExisting=false ~~~ ## 靜態攔截器靜態攔截器的作用是將k/v插入到事件的報頭中。配置如下 ![](https://box.kancloud.cn/b39173374252bc5040ff126caff8856b_695x409.png) source連接到靜態攔截器的配置： ~~~ a1.sources.r1.interceptors=?i3 a1.sources.r1.interceptors.static.type=static # 自己指定k/v a1.sources.r1.interceptors.static.key=logs a1.sources.r1.interceptors.static.value=logFlume a1.sources.r1.interceptors.static.preserveExisting=false ~~~ ## 正則過濾攔截器在日志采集的時候，可能有一些數據是我們不需要的，這樣添加過濾攔截器，可以過濾掉不需要的日志，也可以根據需要收集滿足正則條件的日志。配置如下 ![](https://box.kancloud.cn/b857f5db3638753c351a4be89fad5979_681x411.png) source連接到正則過濾攔截器的配置： ~~~ a1.sources.r1.interceptors=i4 a1.sources.r1.interceptors.i4.type=REGEX_FILTER # 保留內容中出現rm或者kill的字符串的記錄 a1.sources.r1.interceptors.i4.regex=(rm)|(kill) a1.sources.r1.interceptors.i4.excludeEvents=false ~~~ 這樣配置的攔截器就只會接收日志消息中帶有rm 或者kill的日志。測試案例： test_regex.conf ~~~ # 定義這個agent中各組件的名字 a1.sources = r1 a1.sinks = k1 a1.channels = c1 # 描述和配置source組件：r1 a1.sources.r1.type = netcat a1.sources.r1.bind = master a1.sources.r1.port = 44444 a1.sources.r1.interceptors=i4 a1.sources.r1.interceptors.i4.type=REGEX_FILTER #保留內容中出現hadoop或者是spark的字符串的記錄 a1.sources.r1.interceptors.i4.regex=(hadoop)|(spark) a1.sources.r1.interceptors.i4.excludeEvents=false # 描述和配置sink組件：k1 # 輸出到控制臺上 a1.sinks.k1.type = logger # 描述和配置channel組件，此處使用是內存緩存的方式 a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100 # 描述和配置source channel sink之間的連接關系 a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1 ~~~ 只接受存在hadoop或者spark記錄