# Flume业务模型配置说明
#### 业务模型配置指导
本章节适用于MRS 3.x及之后版本。
本任务旨在提供Flume常用模块的性能差异，用于指导用户进行合理的Flume业务配置，避免出现前端Source和后端Sink性能不匹配进而导致整体业务性能不达标的场景。
本任务只针对于单通道的场景进行比较说明。
Flume业务配置及模块选择过程中，一般要求Sink的极限吞吐量需要大于Source的极限吞吐量，否则在极限负载的场景下，Source往Channel的写入速度大于Sink从Channel取出的速度，从而导致Channel频繁被写满，进而影响性能表现。
Avro Source和Avro Sink一般都是成对出现，用于多个Flume Agent间进行数据中转，因此一般场景下Avro Source和Avro Sink都不会成为性能瓶颈。
#### 模块间性能
根据模块间性能对比，可以看到对于前端是SpoolDir Source的场景下，Kafka Sink和HDFS Sink都能满足吞吐量要求，但是HBase Sink由于自身写入性能较低的原因，会成为性能瓶颈，会导致数据都积压在Channel中。但是如果有必须使用HBase Sink或者其他性能容易成为瓶颈的Sink的场景时，可以选择使用**Channel Selector** 或者**Sink Group**来满足性能要求。
#### Channel Selector
Channel Selector可以允许一个Source对接多个Channel，通过选择不同的Selector类型来将Source的数据进行分流或者复制，目前Flume提供的Channel Selector有两种：Replicating和Multiplexing。
Replicating：表示Source的数据同步发送给所有Channel。
Multiplexing：表示根据Event中的Header的指定字段的值来进行判断，从而选择相应的Channel进行发送，从而起到根据业务类型进行分流的目的。
- Replicating配置样例：
  ```
  client.sources = kafkasource
  client.channels = channel1 channel2
  client.sources.kafkasource.type = org.apache.flume.source.kafka.KafkaSource
  client.sources.kafkasource.kafka.topics = topic1,topic2
  client.sources.kafkasource.kafka.consumer.group.id = flume
  client.sources.kafkasource.kafka.bootstrap.servers = 10.69.112.108:21007
  client.sources.kafkasource.kafka.security.protocol = SASL_PLAINTEXT
  client.sources.kafkasource.batchDurationMillis = 1000
  client.sources.kafkasource.batchSize = 800
  client.sources.kafkasource.channels = channel1 channel2
  client.sources.kafkasource.selector.type = replicating
  client.sources.kafkasource.selector.optional = channel2
  ```
  表1Replicating配置样例参数说明 
  | 选项名称              | 默认值         | 描述                         |
  |:---|:---|:---|
  | Selector.type     | replicating | Selector类型，应配置为replicating |
  | Selector.optional | -           | 可选Channel，可以配置为列表          |
     
  
- Multiplexing配置样例：
  ```
  client.sources = kafkasource
  client.channels = channel1 channel2
  client.sources.kafkasource.type = org.apache.flume.source.kafka.KafkaSource
  client.sources.kafkasource.kafka.topics = topic1,topic2
  client.sources.kafkasource.kafka.consumer.group.id = flume
  client.sources.kafkasource.kafka.bootstrap.servers = 10.69.112.108:21007
  client.sources.kafkasource.kafka.security.protocol = SASL_PLAINTEXT
  client.sources.kafkasource.batchDurationMillis = 1000
  client.sources.kafkasource.batchSize = 800
  client.sources.kafkasource.channels = channel1 channel2
  client.sources.kafkasource.selector.type = multiplexing
  client.sources.kafkasource.selector.header = myheader
  client.sources.kafkasource.selector.mapping.topic1 = channel1
  client.sources.kafkasource.selector.mapping.topic2 = channel2
  client.sources.kafkasource.selector.default = channel1
  ```
  表2Multiplexing配置样例参数说明 
  | 选项名称                | 默认值                   | 描述                          |
  |:---|:---|:---|
  | Selector.type       | replicating           | Selector类型，应配置为multiplexing |
  | Selector.header     | Flume.selector.header | -                           |
  | Selector.default    | -                     | -                           |
  | Selector.mapping.\* | -                     | -                           |
     
  Multiplexing类型的Selector的样例中，选择Event中Header名称为topic的字段来进行判断，当Header中topic字段的值为topic1时，向channel1发送该Event，当Header中topic字段的值为topic2时，向channel2发送该Event。
  这种Selector需要借助Source中Event的特定Header来进行Channel的选择，需要根据业务场景选择合理的Header来进行数据分流。
  
 
#### SinkGroup
当后端单Sink性能不足、需要高可靠性保证或者异构输出时可以使用Sink Group来将指定的Channel和多个Sink对接，从而满足相应的使用场景。目前Flume提供了两种Sink Processor用于对Sink Group中的Sink进行管理：Load Balancing和Failover。
Failover：表示在Sink Group中同一时间只有一个Sink处于活跃状态，其他Sink作为备份处于非活跃状态，当活跃状态的Sink故障时，根据优先级从非活跃状态的Sink中选择一个来接管业务，保证数据不会丢失，多用于高可靠性场景。
Load Balancing：表示在Sink Group中所有Sink都处于活跃状态，每个Sink都会从Channel中去获取数据并进行处理，并且保证在运行过程中该Sink Group的所有Sink的负载是均衡的，多用于性能提升场景。
- Load Balancing配置样例：
  ```
  client.sources = source1  
  client.sinks = sink1 sink2
  client.channels = channel1
  client.sinkgroups = g1
  client.sinkgroups.g1.sinks = sink1 sink2
  client.sinkgroups.g1.processor.type = load_balance
  client.sinkgroups.g1.processor.backoff = true
  client.sinkgroups.g1.processor.selector = random
  client.sinks.sink1.type = logger
  client.sinks.sink1.channel = channel1
  client.sinks.sink2.type = logger
  client.sinks.sink2.channel = channel1
  ```
  表3Load Balancing配置样例参数说明 
  | 选项名称                          | 默认值         | 描述                                                           |
  |:---|:---|:---|
  | sinks                         | -           | Sink Group的sink列表，多个以空格分隔                                    |
  | processor.type                | default     | Processor的类型，应配置为load_balance                                |
  | processor.backoff             | false       | 是否以指数的形式退避失败的Sinks                                           |
  | processor.selector            | round_robin | 选择机制。必须是round_robin，random或者自定义的类，且该类继承了AbstractSinkSelector |
  | processor.selector.maxTimeOut | 30000       | 屏蔽故障sink的时间，默认是30000毫秒                                       |
     
  
- Failover配置样例：
  ```
  client.sources = source1       
  client.sinks = sink1 sink2
  client.channels = channel1
  client.sinkgroups = g1
  client.sinkgroups.g1.sinks = sink1 sink2
  client.sinkgroups.g1.processor.type = failover
  client.sinkgroups.g1.processor.priority.sink1 = 10
  client.sinkgroups.g1.processor.priority.sink2 = 5
  client.sinkgroups.g1.processor.maxpenalty = 10000
  client.sinks.sink1.type = logger
  client.sinks.sink1.channel = channel1
  client.sinks.sink2.type = logger
  client.sinks.sink2.channel = channel1
  ```
  表4**Failover** 配置样例参数说明 
  | 选项名称                            | 默认值     | 描述                                                                                                     |
  |:---|:---|:---|
  | sinks                           | -       | Sink Group的sink列表，多个以空格分隔                                                                              |
  | processor.type                  | default | Processor的类型，应配置为failover                                                                              |
  | processor.priority.\<sinkName\> | -       | 优先级值。\<sinkName\> 必须是sinks中有定义的。优先级值高Sink会更早被激活。值越大，优先级越高。**注**：多个sinks的话，优先级的值不要相同，如果优先级相同的话，只会有一个生效。 |
  | processor.maxpenalty            | 30000   | 失败的Sink最大的退避时间(单位：毫秒)                                                                                  |
     
  
 
#### Interceptors
Flume的拦截器（Interceptor）支持在数据传输过程中修改或丢弃传输的基本单元Event。用户可以通过在配置中指定Flume内建拦截器的类名列表，也可以开发自定义的拦截器来实现Event的修改或丢弃。Flume内建支持的拦截器如下表所示，本章节会选取一个较为复杂的作为示例。其余的用户可以根据需要自行配置使用。官网参考：<https://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html>
![](https://support.huaweicloud.com/cmpntguide-mrs/public_sys-resources/note_3.0-zh-cn.png)
1. 拦截器用在Flume的Source、Channel之间，大部分的Source都带有Interceptor参数。用户可以依据需要配置。
2. Flume支持一个Source配置多个拦截器，各拦截器名称用空格分开。
3. 指定拦截器的顺序就是它们被调用的顺序。
4. 使用拦截器在Header中插入的内容，都可以在Sink中读取并使用。
 
表5Flume内建支持的拦截器类型 
| 拦截器类型                          | 简要描述                                                                    |
|:---|:---|
| Timestamp Interceptor          | 该拦截器会在Event的Header中插入一个时间戳。                                             |
| Host Interceptor               | 该拦截器会在Event的Header中插入当前Agent所在节点的IP或主机名。                                |
| Remove Header Interceptor      | 该拦截器会依据Header中包含的符合正则匹配的字符串，丢弃掉对应的Event。                                |
| UUID Interceptor               | 该拦截器会为每个Event的Header生成一个UUID字符串。                                        |
| Search and Replace Interceptor | 该拦截器基于Java正则表达式提供简单的基于字符串的搜索和替换功能。与Java Matcher.replaceAll() 的规则相同。     |
| Regex Filtering Interceptor    | 该拦截器通过将Event的Body体解释为文本文件，与配置的正则表达式进行匹配来选择性的过滤Event。提供的正则表达式可用于排除或包含事件。 |
| Regex Extractor Interceptor    | 该拦截器使用正则表达式抽取原始events中的内容，并将该内容加入events的header中。                        |
   
下面以Regex Filtering Interceptor 为例说明Interceptor使用（其余的可参考官网配置）：
表6Regex Filtering Interceptor配置参数说明 
| 选项名称          | 默认值   | 描述                                         |
|:---|:---|:---|
| type          | -     | 组件类型名称，必须写为regex_filter。                   |
| regex         | -     | 用于匹配事件的正则表达式。                              |
| excludeEvents | false | 默认收集匹配到的Event。设置为true，则会删除匹配的Event，保留不匹配的。 |
   
配置示例（为了方便观察，此模型使用了netcat tcp作为Source源，logger作为Sink）。配置好如下参数后，在Linux的配置的主机节点上执行Linux命令"telnet *主机名或IP* 44444"，并任意敲入符合正则和不符合正则的字符串。会在日志中观察到，只有匹配到的字符串被传输了。
```
#define the source、channel、sink
server.sources = r1
server.channels = c1
server.sinks = k1
#config the source
server.sources.r1.type = netcat
server.sources.r1.bind = ${主机IP}
server.sources.r1.port = 44444
server.sources.r1.interceptors= i1
server.sources.r1.interceptors.i1.type= regex_filter
server.sources.r1.interceptors.i1.regex= (flume)|(myflume)
server.sources.r1.interceptors.i1.excludeEvents= false
server.sources.r1.channels = c1
#config the channel
server.channels.c1.type = memory
server.channels.c1.capacity = 1000
server.channels.c1.transactionCapacity = 100
#config the sink
server.sinks.k1.type = logger
server.sinks.k1.channel = c1
```