flume中的拦截器

Posted 2021-01-16 beiyi888

Flume中的拦截器（interceptor），用户Source读取events发送到Sink的时候，在events header中加入一些有用的信息，或者对events的内容进行过滤，完成初步的数据清洗。这在实际业务场景中非常有用，Flume-ng 1.6中目前提供了以下拦截器：

Timestamp Interceptor；
Host Interceptor；
Static Interceptor；
UUID Interceptor；
Morphline Interceptor；
Search and Replace Interceptor；
Regex Filtering Interceptor；
Regex Extractor Interceptor；

本文对常用的几种拦截器进行学习和介绍，并附上使用示例。

对一个Source可以使用多个拦截器。

Timestamp Interceptor

时间戳拦截器，将当前时间戳（毫秒）加入到events header中，key名字为：timestamp，值为当前时间戳。用的不是很多。比如在使用HDFS Sink时候，根据events的时间戳生成结果文件，hdfs.path = hdfs://cdh5/tmp/dap/%Y%m%d

hdfs.filePrefix = log_%Y%m%d_%H

会根据时间戳将数据写入相应的文件中。

但可以用其他方式代替（设置useLocalTimeStamp = true）。

Host Interceptor

主机名拦截器。将运行Flume agent的主机名或者IP地址加入到events header中，key名字为：host（也可自定义）。

Static Interceptor

静态拦截器，用于在events header中加入一组静态的key和value。

UUID Interceptor

UUID拦截器，用于在每个events header中生成一个UUID字符串，例如：b5755073-77a9-43c1-8fad-b7a586fc1b97。生成的UUID可以在sink中读取并使用。