Flume NG 學習筆記（八）Interceptors（攔截器）測試
攔截器主要是對事件的header資訊資訊操作，要麼直接忽略他，要麼修改他的資料
一、Event Serializers
file_roll sink 和hdfs sink 都支援EventSerializer介面
1.1、Body Text Serializer
Body TextSerializer，別名：text。這個攔截器將把事件的body部分寫入到輸出流中而不需要任何轉換或者修改。事件的header將直接被忽略。
下面是官網配置：
|
Property Name |
Default |
Description |
|
appendNewline |
true |
Whether a newline will be appended to each event at write time. The default of true assumes that events do not contain newlines, for legacy reasons. |
下面是官網例子：appendNewline是選擇是否加入到新行去。預設是true，而false 就是換行，一般我們都選擇換行。
a1.sinks=k1
a1.sinks.k1.type=file_roll
a1.sinks.k1.channel=c1
a1.sinks.k1.sink.directory=/var/log/flume
a1.sinks.k1.sink.serializer=text
a1.sinks.k1.sink.serializer.appendNewline=false
下面是實際例子
因為是考慮Body TextSerializer的特性，他會忽略header的資訊，因此我們這邊要採用http source來接收定義的header 與body 的內容
#配置檔案：body_case15.conf # Name the components on this agent a1.sources = r1 a1.sinks = k1 a1.channels = c1 # Describe/configure the source a1.sources.r1.type = http a1.sources.r1.port = 50000 a1.sources.r1.host = 192.168.233.128 a1.sources.r1.channels = c1 # Describe the sink a1.sinks.k1.type = file_roll a1.sinks.k1.channel = c1 a1.sinks.k1.sink.directory = /tmp/logs a1.sinks.k1.sink.serializer = text a1.sinks.k1.sink.serializer.appendNewline =false # Use a channel which buffers events inmemory a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100
#敲命令
flume-ng agent -c conf -fconf/body_case15.conf -n a1 -Dflume.root.logger=INFO,console
啟動成功後
開啟另一個終端輸入，往偵聽埠送資料
curl -X POST -d '[{"headers":{"looklook1" : "looklook1 isheader","looklook2": "looklook2 isheader"},"body" : "hellolooklook5"}]'http://192.168.233.128:50000
#在啟動源傳送的代理終端檢視console輸出
資料已經輸出，但只輸出了hello looklook5，即BODY這塊。
1.2、Avro Event Serializer
Avro Event Serializer別名：avro_event。這個攔截器將把事件序列化到一個Avro容器檔案中。使用的模式和RPC Avro機制使用到的處理flume事件的機制一樣。這個序列化器繼承自AbstractAvroEventSerializer類。
官網例子
|
Property Name |
Default |
Description |
|
syncIntervalBytes |
2048000 |
Avro sync interval, in approximate bytes. |
|
compressionCodec |
null |
Avro compression codec. For supported codecs, see Avro’s CodecFactory docs. |
下面是官網例子
a1.sinks.k1.type=hdfs
a1.sinks.k1.channel=c1
a1.sinks.k1.hdfs.path=/flume/events/%y-%m-%d/%H%M/%S
a1.sinks.k1.serializer=avro_event
a1.sinks.k1.serializer.compressionCodec=snappy
例子這邊就不演示了，因為和BodyText Serializer 差距不大。
二、Timestamp Interceptor
官網說Flume 可以在事件傳輸過程中對它進行修改與刪除，而這個都是通過Interceptor進行實現的，實際都是往事件的header裡插資料。而Timestamp Interceptor攔截器就是可以往event的header中插入關鍵詞為timestamp的時間戳。
下面是官網配置
|
Property Name |
Default |
Description |
|
type |
– |
The component type name, has to be timestamp or the FQCN |
|
preserveExisting |
false |
If the timestamp already exists, should it be preserved - true or false |
以及官網例子
a1.sources=r1
a1.channels=c1
a1.sources.r1.channels= c1
a1.sources.r1.type=seq
a1.sources.r1.interceptors=i1
a1.sources.r1.interceptors.i1.type=timestamp
下面是測試例子
#配置檔案：timestamp_case16.conf
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = syslogtcp
a1.sources.r1.port = 50000
a1.sources.r1.host = 192.168.233.128
a1.sources.r1.channels = c1
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.preserveExisting= false
a1.sources.r1.interceptors.i1.type = timestamp
# Describe the sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.channel = c1
a1.sinks.k1.hdfs.path =hdfs://carl:9000/flume/%Y-%m-%d/%H%M
a1.sinks.k1.hdfs.filePrefix = looklook5.
a1.sinks.k1.hdfs.fileType=DataStream
# Use a channel which buffers events inmemory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100這裡拿header作為資料夾目錄名稱。
#敲命令
flume-ng agent -c conf -f conf/timestamp_case16.conf-n a1 -Dflume.root.logger=INFO,console
啟動成功後
開啟另一個終端輸入，往偵聽埠送資料
echo "TimestampInterceptor" | nc 192.168.233.128 50000
#在啟動源傳送的代理終端檢視console輸出
檢視hdfs生成的檔案,可以看到timestamp已經生成在header裡面,可以根據自定義的格式生成資料夾，資料也都傳輸過來了。
三、Host Interceptor
該攔截器可以往event的header中插入關鍵詞預設為host主機名或者ip地址（注意是agent執行的機器的主機名或者ip地址）
下面是官網配置
|
Property Name |
Default |
Description |
|
type |
– |
The component type name, has to be host |
|
preserveExisting |
false |
If the host header already exists, should it be preserved - true or false |
|
useIP |
true |
Use the IP Address if true, else use hostname. |
|
hostHeader |
host |
The header key to be used. |
以及官網例子
a1.sources=r1
a1.channels=c1
a1.sources.r1.interceptors=i1
a1.sources.r1.interceptors.i1.type=host
a1.sources.r1.interceptors.i1.hostHeader=hostname
下面是測試例子
#配置檔案：time_host_case17.conf
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = syslogtcp
a1.sources.r1.port = 50000
a1.sources.r1.host = 192.168.233.128
a1.sources.r1.channels = c1
a1.sources.r1.interceptors = i1 i2
a1.sources.r1.interceptors.i1.preserveExisting= false
a1.sources.r1.interceptors.i1.type =timestamp
a1.sources.r1.interceptors.i2.type = host
a1.sources.r1.interceptors.i2.hostHeader =hostname
a1.sources.r1.interceptors.i2.useIP = false
# Describe the sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.channel = c1
a1.sinks.k1.hdfs.path =hdfs://carl:9000/flume/%Y-%m-%d/%H%M
a1.sinks.k1.hdfs.filePrefix = %{hostname}
a1.sinks.k1.hdfs.fileType=DataStream
# Use a channel which buffers events inmemory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100增加一個攔截器，型別是host,h將hostname作為檔案字首。
#敲命令
flume-ng agent -c conf -f conf/time_host_case17.conf-n a1 -Dflume.root.logger=INFO,console
啟動成功後
開啟另一個終端輸入，往偵聽埠送資料
echo "Time&hostInterceptor1" | nc 192.168.233.128 50000
echo "Time&hostInterceptor2" | nc 192.168.233.128 50000
#在啟動源傳送的代理終端檢視console輸出
檢視hdfs生成的檔案,可以看到host已經生成在header裡面,可以根據自定義的格式生成資料夾，資料也都傳輸過來了。
四、Static Interceptor
Static Interceptor攔截器允許使用者增加一個static的header併為所有的事件賦值。範圍是所有事件。
官網配置
|
Property Name |
Default |
Description |
|
type |
– |
The component type name, has to be static |
|
preserveExisting |
true |
If configured header already exists, should it be preserved - true or false |
|
key |
key |
Name of header that should be created |
|
value |
value |
Static value that should be created |
其中引數key與value等於類似json格式裡的"headers":{" key":" value"}
下面是官網例子
a1.sources=r1
a1.channels=c1
a1.sources.r1.channels= c1
a1.sources.r1.type=seq
a1.sources.r1.interceptors=i1
a1.sources.r1.interceptors.i1.type=static
a1.sources.r1.interceptors.i1.key=datacenter
a1.sources.r1.interceptors.i1.value=NEW_YORK
以及實際的列子
#配置檔案：static_case18.conf
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = syslogtcp
a1.sources.r1.port = 50000
a1.sources.r1.host = 192.168.233.128
a1.sources.r1.channels = c1
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = static
a1.sources.r1.interceptors.i1.key = looklook5
a1.sources.r1.interceptors.i1.value =looklook10
# Describe the sink
a1.sinks.k1.type = logger
# Use a channel which buffers events inmemory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1#敲命令
flume-ng agent -c conf -f conf/static_case18.conf-n a1 -Dflume.root.logger=INFO,console
啟動成功後
開啟另一個終端輸入，往偵聽埠送資料
echo "statInterceptor1" | nc 192.168.233.128 50000
#在啟動源傳送的代理終端檢視console輸出
可以看到輸出的header資訊裡自定義部分正確輸出，body部分也輸出正確。
五、Regex FilteringInterceptor
Regex Filtering Interceptor攔截器用於過濾事件，篩選出與配置的正則表示式相匹配的事件。可以用於包含事件和排除事件。常用於資料清洗，通過正則表示式把資料過濾出來。
官網配置
|
Property Name |
Default |
Description |
|
type |
– |
The component type name has to be regex_filter |
|
regex |
”.*” |
Regular expression for matching against events |
|
excludeEvents |
false |
If true, regex determines events to exclude, otherwise regex determines events to include. |
excludeEvents 為true的時候為排除所有匹配正則表示式的資料。
下面是測試例子
#配置檔案：regex_filter_case19.conf
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = syslogtcp
a1.sources.r1.port = 50000
a1.sources.r1.host = 192.168.233.128
a1.sources.r1.channels = c1
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type =regex_filter
a1.sources.r1.interceptors.i1.regex =^[0-9]*$
a1.sources.r1.interceptors.i1.excludeEvents =true
# Describe the sink
a1.sinks.k1.type = logger
# Use a channel which buffers events inmemory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1我們對開頭字母是數字的資料，全部過濾。
#敲命令
flume-ng agent -c conf -f conf/regex_filter_case19.conf-n a1 -Dflume.root.logger=INFO,console
啟動成功後
開啟另一個終端輸入，往偵聽埠送資料
echo "a" | nc192.168.233.128 50000
echo "1222" |nc 192.168.233.128 50000
echo "a222" |nc 192.168.233.128 50000
#在啟動源傳送的代理終端檢視console輸出
可以看出1222 被認為是無效的資料沒有發出來。
Regex Filtering Interceptor測試成功。