Flume - 故障转移、负载均衡

快来打我* 2023-06-09 10:27 1阅读 0赞

*  故障转移

概述：

> 故障转移机制的工作方式是将失败的sink放到一个池中，并在池中为它们分配一段冷冻期，在重试之前随着连续的失败而增加。一个sink成功发送event后，将其恢复到活动池。sink有一个与它们相关联的优先级，数字越大表示优先级越高。如果一个sink在发送event时失败，则下一个具有最高优先级的sink将被尝试用于发送事件。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0JsZXNzaW5nWFJZ_size_16_color_FFFFFF_t_70][]

实例：

# hadoop105
    # Name the components on this agent
    a1.sources = r1
    a1.sinks = k1 k2
    a1.channels = c1
    a1.sinkgroups = g1
    
    # Describe/configure the source
    a1.sources.r1.type = netcat
    a1.sources.r1.bind = localhost
    a1.sources.r1.port = 44444
    
    # Channel
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    # Sink
    a1.sinks.k1.type = avro
    a1.sinks.k1.hostname = hadoop106
    a1.sinks.k1.port = 4141
    
    a1.sinks.k2.type = avro
    a1.sinks.k2.hostname = hadoop107
    a1.sinks.k2.port = 4142
    
    #Sink Group
    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
    
    #Bind
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1
    a1.sinks.k2.channel = c1
    ----------------------------------------------------------------------
    # hadoop106
    #Name
    a2.sources = r1
    a2.channels = c1
    a2.sinks = k1
    
    #Source
    a2.sources.r1.type = avro
    a2.sources.r1.bind = hadoop106
    a2.sources.r1.port = 4141
    
    #Channel
    a2.channels.c1.type = memory
    a2.channels.c1.capacity = 1000
    a2.channels.c1.transactionCapacity = 100
    
    #Sink
    a2.sinks.k1.type = logger
    
    #Bind
    a2.sources.r1.channels = c1
    a2.sinks.k1.channel = c1
    ----------------------------------------------------------------------
    # hadoop107
    #Name
    a3.sources = r1
    a3.channels = c1
    a3.sinks = k1
    
    #Source
    a3.sources.r1.type = avro
    a3.sources.r1.bind = hadoop107
    a3.sources.r1.port = 4142
    
    #Channel
    a3.channels.c1.type = memory
    a3.channels.c1.capacity = 1000
    a3.channels.c1.transactionCapacity = 100
    
    #Sink
    a3.sinks.k1.type = logger
    
    #Bind
    a3.sources.r1.channels = c1
    a3.sinks.k1.channel = c1

> 参考文章链接：  
> 文章1：http://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html  
> 文章2：https://blog.51cto.com/jackwxh/1906893

*  负载均衡

概述：

> source中的event流经channel，进入sink group，在sink group中根据负载算法（round\_robin、random）选择sink，然后选择不同机器上的agent实现负载均衡。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0JsZXNzaW5nWFJZ_size_16_color_FFFFFF_t_70 1][]

# hadoop105
    # Name the components on this agent
    a1.sources = r1
    a1.sinks = k1 k2
    a1.channels = c1
    a1.sinkgroups = g1
    
    # Describe/configure the source
    a1.sources.r1.type = netcat
    a1.sources.r1.bind = localhost
    a1.sources.r1.port = 44444
    
    # Channel
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    # Sink
    a1.sinks.k1.type = avro
    a1.sinks.k1.hostname = hadoop106
    a1.sinks.k1.port = 4141
    
    a1.sinks.k2.type = avro
    a1.sinks.k2.hostname = hadoop107
    a1.sinks.k2.port = 4142
    
    #Sink Group
    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
    
    #Bind
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1
    a1.sinks.k2.channel = c1
    ----------------------------------------------------------------------
    # hadoop106
    #Name
    a2.sources = r1
    a2.channels = c1
    a2.sinks = k1
    
    #Source
    a2.sources.r1.type = avro
    a2.sources.r1.bind = hadoop106
    a2.sources.r1.port = 4141
    
    #Channel
    a2.channels.c1.type = memory
    a2.channels.c1.capacity = 1000
    a2.channels.c1.transactionCapacity = 100
    
    #Sink
    a2.sinks.k1.type = logger
    
    #Bind
    a2.sources.r1.channels = c1
    a2.sinks.k1.channel = c1
    ----------------------------------------------------------------------
    # hadoop107
    #Name
    a3.sources = r1
    a3.channels = c1
    a3.sinks = k1
    
    #Source
    a3.sources.r1.type = avro
    a3.sources.r1.bind = hadoop107
    a3.sources.r1.port = 4142
    
    #Channel
    a3.channels.c1.type = memory
    a3.channels.c1.capacity = 1000
    a3.channels.c1.transactionCapacity = 100
    
    #Sink
    a3.sinks.k1.type = logger
    
    #Bind
    a3.sources.r1.channels = c1
    a3.sinks.k1.channel = c1

> 参考文章地址：  
> 文章1：http://flume.apache.org/releases/content/1.9.0/FlumeUserGuide.html  
> 文章2：https://blog.csdn.net/silentwolfyh/article/details/51165804

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0JsZXNzaW5nWFJZ_size_16_color_FFFFFF_t_70]: /images/20230531/91a78aceba2e462aaf38c72374bf153f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0JsZXNzaW5nWFJZ_size_16_color_FFFFFF_t_70 1]: /images/20230531/a285cf8f40a1478693bf92e51748835e.png