高阶使用

拦截器的位置在Source和Channel之间，用于修改或丢弃Event。拦截图示意图如下。

拦截器的主要类型如下表。

类型	描述
时间戳拦截器	在Event Header中添加Unix时间戳属性。
Host拦截器	在Event Header中添加Host属性。
静态拦截器	在Event Header中添加一个固定键值对属性。
Header拦截器	在Event Header中删除一个或多个属性。
UUID拦截器	在Event中设置一个UUID。如果应用层没有UUID，则可以使用该拦截器来默认添加。
Morphline拦截器	通过Morphline配置文件过滤Event或修改插入Event Header。
查找拦截器	使用Java正则表达式查找Event Body。
替换拦截器	使用Java正则表达式替换Event Body。
正则过滤拦截器	过滤配置匹配或者没有匹配上正则表达式的Event Body。

相关示例如下：

• 示例1：Event Body包含1:2:3.4foobar5，如果想配置正则过滤器，则配置如下。

  a1.sources.r1.interceptors.i1.regex = (\\d):(\\d):(\\d)
  a1.sources.r1.interceptors.i1.serializers = s1 s2 s3
  a1.sources.r1.interceptors.i1.serializers.s1.name = one
  a1.sources.r1.interceptors.i1.serializers.s2.name = two
  a1.sources.r1.interceptors.i1.serializers.s3.name = three

  修改后的Event Body不变，后续的Header中增加了one=>1, two=>2, three=>3。

• 示例2：Event Body包含2012-10-18 18:47:57,614 some log line，如果想配置时间过滤器，则配置如下。

  a1.sources.r1.interceptors.i1.regex = ^(?:\\n)?(\\d\\d\\d\\d-\\d\\d-\\d\\d\\s\\d\\d:\\d\\d)
  a1.sources.r1.interceptors.i1.serializers = s1
  a1.sources.r1.interceptors.i1.serializers.s1.type = org.apache.flume.interceptor.RegexExtractorInterceptorMillisSerializer
  a1.sources.r1.interceptors.i1.serializers.s1.name = timestamp
  a1.sources.r1.interceptors.i1.serializers.s1.pattern = yyyy-MM-dd HH:mm

  修改后的Event Body不变，后续的Header中增加了timestamp=>1350611220000。

Channel选择器用于在Source与Channel一对多场景下选择Channel。Channel选择器示意图如下。

Flume内置复制选择器（Replicating）和多路复用选择器（Multiplexing）两种选择器，默认为复制选择器。复制选择器会把所有Event发送到每个Channel，而多路复用选择器，则会按照一定的规则发送。多路复用选择器的示例如下。

a1.sources = r1
a1.channels = c1 c2 c3 c4
a1.sources.r1.selector.type = multiplexing
a1.sources.r1.selector.header = state
a1.sources.r1.selector.mapping.CZ = c1
a1.sources.r1.selector.mapping.US = c2 c3
a1.sources.r1.selector.default = c4

Sink组逻辑处理器用于多个Sink一同消费Channel队列中的数据，并把这些Sink配置为负载均衡或故障转移的工作方式。默认Sink与Channel是一对一的。配置为负载均衡方式，则根据配置的负载均衡机制，将Event分发到Sink中。配置为故障转移方式，则表示多个Sink是一主多备的工作方式，当工作的Sink中止后，Event会被转移到备用的Sink上。

相关示例如下：

• 示例1：故障转移方式

  a1.sinkgroups = g1
  a1.sinkgroups.g1.sinks = k1 k2
  a1.sinkgroups.g1.processor.type = failover
  a1.sinkgroups.g1.processor.priority.k1 = 5
  a1.sinkgroups.g1.processor.priority.k2 = 10
  a1.sinkgroups.g1.processor.maxpenalty = 10000

  上述示例中有k1和k2两个Sink，权重分别是5和10，最大故障转移时间是10000毫秒。

• 示例2：负载均衡方式

  a1.sinkgroups = g1
  a1.sinkgroups.g1.sinks = k1 k2
  a1.sinkgroups.g1.processor.type = load_balance
  a1.sinkgroups.g1.processor.backoff = true
  a1.sinkgroups.g1.processor.selector = random

  上述示例中有k1和k2两个Sink，通过随机选择（random）方法进行负载分配，您也可以使用轮询（round_robin）方法。示例中的a1.sinkgroups.g1.processor.backoff参数表示是否以指数的形式退避失败的Sinks，设置为true，则Sink Processor会屏蔽故障的Sink。