Flume之多层代理

矫情吗；* 2022-11-28 13:39 132阅读 0赞

## 为什么需要多层代理 ##

当我们用 Flume 采集日志时，由于数据源的多样性，则往往需要配置多个Flume 进行采集，如果只是使用单层 Flume 的话，那么往往会产生很多个文件夹，单个文件夹也只是来自同一个节点的数据组成的。

而实际开发中，为了减少HDFS 的压力，同时提高后续 MR 的处理效率。往往会将同一组多个节点的数据汇聚到同一个文件中，这样同时也较少了数据从生产到分析的时间。

如下图，第一次 agent 负责采集原始数据，第二层 agent 负责对第一层数据进行汇聚。这种多层代理的方式尤其适合 source 源数据量庞大的时候，效率会高很多。  
![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center][]

## 多路复用流程 ##

Flume支持复用event流向一个或多个目的地。通过定义可复制或者选择性地将event路由到一个或多个channels的流复用器实现。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center 1]  
上述例子展示了一个agent "foo"的source把流量分散到三个不同的channels。这分散可以被复制或复用。在复制流量的情况下，每个event都被发送到三个channels。  
对于复用情况，当一个event的属性匹配到预配置的值时，该event被传递到可用channels的子集。

例如，如果一个event属性“txnType” 被设置为“customer”，然后它应该去channel1和channel3，如果它是“vendor”，然后它应该去channel2，否则channel3。这个映射可以设置在agent的配置文件中。

如果没有手动配置，source 的默认 channel 选择器类型是 replicating（复制），当然这个选择器只针对 source 配置了多个 channel 的时候。  
前面介绍过，一个source 可以向多个 channel 同时写数据，所以也就产生了以何种方式向多个  
channel 写的问题（比如自带的复制选择器 ，会把数据完整地发送到每一个channel，而多路复用选择器 就可以通过配置来按照一定的规则进行分发，听起来很像负载均衡），channel 选择器也就应运而生。

#### 1. 复制选择器 ####

a1.sources = r1
    a1.channels = c1 c2 c3
    a1.sources.r1.selector.type = replicating
    a1.sources.r1.channels = c1 c2 c3
    
    #向 c3 发送失败将忽略
    a1.sources.r1.selector.optional = c3

上面这个例子中，c3 配置成了可选的。向 c3 发送数据如果失败了会被忽略。  
c1 和 c2 没有配置成可选的，向 c1 和 c2 写数据失败会导致事务失败回滚。

#### 2. 多路复用选择器 ####

a1.sources = r1
    a1.channels = c1 c2 c3 c4
    a1.sources.r1.selector.type = multiplexing
    #以每个 Event 的 header 中的 state 这个属性的值作为选择 channel 的依据
    a1.sources.r1.selector.header = state 
    
    a1.sources.r1.selector.mapping.CZ = c1 
    #如果 state=US，则选择 c2 和 c3 这两个 channel 
    a1.sources.r1.selector.mapping.US = c2 c3 
    #默认使用 c4 这个 channel，如果没有被规则匹配到，默认会发到此 channel
    a1.sources.r1.selector.default = c4

## 分散流程 ##

Flume支持从一个source到多个channels的分散流程。分散有两种模式，复制(replicating)和复用(multiplexing)。在复制流程中，event将会被发送到全部配置的channels。  
在复用情况下，event只会被发送到符合要求的channels。

在分散流程中，需要指定source的channels列表和分散规则。添加一个channel “selector”，可以是replicating或者multiplexing，然后指定选择规则，如果你没有指定一个selector，默认replicating

a1.sources = s1
    a1.channels = c1 c2
    a1.sinks = k1 k2
    
    a1.sources.s1.channels = c1 c2
    
    a1.sinks.k1.channel = c1
    a1.sinks.k2.channel = c2
    a1.sources.s1.selector.type = replicating

复用选择有一系列属性去分流流程。这要求在channel设置指定event属性的mapping。selector在event的header检查每个配置的属性。  
如果匹配到指定的值，该event会被发送到映射该值的channels。如果没有匹配到，event将会发送到配置好默认的channels.

a1.sources.s1.selector.type = multiplexing
    a1.sources.s1.selector.header = someHeader
    a1.sources.s1.selector.mapping.value1 = c1
    a1.sources.s1.selector.mapping.value2 = c1 c2
    a1.sources.s1.selector.mapping.value3 = c2
    a1.sources.s1.selector.default = c2

下面的例子展示，一个简单流程多路复用到两条路径。名叫agent\_foo的agent有一个简单avro source和两个链接到两个sinks的channels：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center 2]  
selector检查叫“State”的header。如果值为“CA”，然后它所属的event将会被发送到mem-channel-1，如果值为“AZ”则发送到file-channel-2，或者值为“NY”则两个channels都发送。  
如果“State“ header没有设置或者匹配不到三个值中的一个，则event被发送到默认的mem-channel-1。

selector也支持可选的channel。为header指定可选的channels，配置属性‘optional’可以像下面那样使用：  
![在这里插入图片描述][20200818021705711.png_pic_center]  
selector第一次将会尝试向要求的channels发送events，如果这些channels消费events失败，则提交事务失败，事务将会再次向这些channels提交。  
一旦要求的channels消费完这些events，selector将会向可选的channels发送events。如果这些可选channels消费失败，将会被忽略，不会再次提交事务。

**注意：**  
如果header没有要求的channels，events将会写进默认channels，也会试图写进可选的channels。  
如果没有要求channels指定，指定可选的channels将会同样造成event被写进默认channels。

如果没有默认的和要求的channels，selector将会把events试图写进可选channels。在这种情况下，失败将会被忽略。

## Flume 可靠性保证 ##

可以把多个 Sink 分成一个组， 这时候 Sink 组逻辑处理器 可以对这同一个组里的几个 Sink 进行负载均衡或者其中一个 Sink 发生故障后将输出 Event 的任务转移到其他的 Sink 上。

说的直白一些，这 N 个 Sink 本来是要将 Event 输出到对应的 N 个目的地的，通过 Sink 组逻辑处理器 就可以把这 N 个 sink 配置成负载均衡或者故障转移的工作方式（暂时还不支持自定义的）。

负载均衡的方式是把 channel 里面的 Event按照配置的负载机制（比如轮询）分别发送到 Sink 各自对应的目的地；故障转移就是这 N 个 Sink 同一时间只有一个在工作，其余的作为备用，工作的 Sink 挂掉之后备用的 Sink 顶上。

## 故障转移 ##

故障转移组逻辑处理器维护了一个发送 Event 失败的 sink 的列表，保证有一个 sink 是可用的来发送 Event。

故障转移机制的工作原理是将故障 Sink 降级到一个池中，在池中为它们分配冷却期（超时时间），在重试之前随顺序故障而增加。 Sink 成功发送事件后，它将恢复到实时池。

Sink 具有与之相关的优先级，数值越大，优先级越高。 如果在发送 Event 时 Sink 发生故障，会继续尝试下一个具有最高优先级的 Sink。例如，在优先级为 80 的 Sink 之前激活优先级为 100 的 Sink。如果未指定优先级，则根据配置中的顺序来选取。

要使用故障转移选择器，不仅要设置 Sink 组的选择器为 failover，还有为每一个 Sink 设置一个唯一的优先级数值。可以使用 maxpenalty 属性设置故障转移时间的上限（毫秒）。  
示例：

a1.sinkgroups = g1
    a1.sinkgroups.g1.sinks = k1 k2
    
    #Sink 组逻辑器
    a1.sinkgroups.g1.processor.type = failover
    
    
    #组内 sink 的权重值，<sinkName>必须是当前组关联的 sink 之一。数值（绝对值）越高越早被激活
    a1.sinkgroups.g1.processor.priority.k1 = 5
    a1.sinkgroups.g1.processor.priority.k2 = 10
    
    #发生异常的 sink 最大故障转移时间（默认 30000 毫秒）
    a1.sinkgroups.g1.processor.maxpenalty = 10000

## 负载均衡 ##

**负载均衡** Sink 选择器提供了在多个 Sink 上进行负载均衡流量的功能。 它维护一个活动 Sink 列表的索引来实现负载的分配。**默认支持了 轮询（round\_robin) 和 随机（random) 两种选择机制分配负载。**  
**默认是轮询**，可以通过配置来更改。也可以从 AbstractSinkSelector 继承写一个自定义的选择器。

工作时，此选择器使用其配置的选择机制选择下一个 Sink 并调用它。 如果所选Sink无法正常工作，则处理器通过其配置的选择机制选择下一个可用Sink。此实现不会将失败的 Sink 列入黑名单，而是继续乐观地尝试每个可用的 Sink。

如果所有 Sink 调用都失败了，选择器会将故障抛给 Sink 的运行器。

如果 backoff 设置为 true 则启用了退避机制，失败的 Sink 会被放入黑名单，达到一定的超时时间后会自动从黑名单移除。  
如从黑名单出来后Sink仍然失败;则再次进入黑名单而且超时时间会翻倍，以避免在无响应的 Sink 上浪费过长时间。

如果没有启用退避机制，在禁用此功能的情况下，发生 Sink 传输失败后，会将本次负载传给下一个 Sink 继续尝试，因此这种情况下是不均衡的。  
示例：

a1.sinkgroups = g1
    a1.sinkgroups.g1.sinks = k1 k2
    
    #Sink 组逻辑
    a1.sinkgroups.g1.processor.type = load_balance
    
    #失败的 sink 是否成倍地增加退避它的时间。
    a1.sinkgroups.g1.processor.backoff = true
    
    #负载均衡机制，可选值：round_robin （轮询）、 random （随机选择）、「自
    定义选择器的全限定类名」
    
    a1.sinkgroups.g1.processor.selector = random

后面还会有更多干货，记得关注小编！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center]: /images/20221124/fd0ca853ec56485cb229be9be66b66b7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221124/825741e45b82411fa1cbf3875b147ac0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwMTc4MzQ5OTQwNzE_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221124/0dda7970ae1544a5a20931178388cd37.png
[20200818021705711.png_pic_center]: /images/20221124/571904bd4488453983bcd497724e0e4d.png