溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
版權聲明：本文為博主原創文章，未經博主允許不得轉載。
目錄(?)[+]
攔截器主要是對事件的header信息信息操作，要么直接忽略他，要么修改他的數據
file_roll sink 和hdfs sink 都支持EventSerializer接口
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 的內容
[html] view plain copy
#配置文件：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這塊。
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 差距不大。
官網說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
下面是測試例子
[html] view plain copy
#配置文件：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里面,可以根據自定義的格式生成文件夾，數據也都傳輸過來了。
該攔截器可以往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
下面是測試例子
[html] view plain copy
#配置文件：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的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
以及實際的列子
[html] view plain copy
#配置文件：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 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的時候為排除所有匹配正則表達式的數據。
下面是測試例子
[html] view plain copy
#配置文件：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測試成功。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
