Flume 案例实操 - 单数据源多出口案例(二)

「爱情、让人受尽委屈。」 2022-01-30 13:01 315阅读 0赞

单Source、Channel多Sink(负载均衡)

![10][]

## 需求分析 ##

1.  使用flume-1监控文件变动
2.  flume-1将变动内容传递给flume-2，flume-2负责输出到控制台
3.  flume-1将变动内容传递给flume-3，flume-3负责输出到控制台

![单数据源多出口案例（二）][5ce12370e4b020998ae00b0b.png]

## 实现步骤 ##

### 准备工作 ###

在`/opt/module/flume-1.7.0/job`目录下创建`group2`文件夹

mkdir group2
    cd group2/

### 创建配置文件 ###

#### 创建flume-netcat-flume.conf ####

vim flume-netcat-flume.conf

添加如下内容

# Name the components on this agent
    a1.sources = r1
    a1.channels = c1
    a1.sinkgroups = g1
    a1.sinks = k1 k2
    
    # Describe/configure the source
    a1.sources.r1.type = netcat
    a1.sources.r1.bind = localhost
    a1.sources.r1.port = 44444
    
    a1.sinkgroups.g1.processor.type = load_balance
    a1.sinkgroups.g1.processor.backoff = true
    a1.sinkgroups.g1.processor.selector = round_robin
    a1.sinkgroups.g1.processor.selector.maxTimeOut=10000
    
    # Describe the sink
    a1.sinks.k1.type = avro
    a1.sinks.k1.hostname = hadoop102
    a1.sinks.k1.port = 4141
    
    a1.sinks.k2.type = avro
    a1.sinks.k2.hostname = hadoop102
    a1.sinks.k2.port = 4142
    
    # Describe the channel
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    # Bind the source and sink to the channel
    a1.sources.r1.channels = c1
    a1.sinkgroups.g1.sinks = k1 k2
    a1.sinks.k1.channel = c1
    a1.sinks.k2.channel = c1

#### 创建flume-flume1.conf ####

vim flume-flume1.conf

# Name the components on this agent
    a2.sources = r1
    a2.sinks = k1
    a2.channels = c1
    
    # Describe/configure the source
    a2.sources.r1.type = avro
    a2.sources.r1.bind = hadoop102
    a2.sources.r1.port = 4141
    
    # Describe the sink
    a2.sinks.k1.type = logger
    
    # Describe the channel
    a2.channels.c1.type = memory
    a2.channels.c1.capacity = 1000
    a2.channels.c1.transactionCapacity = 100
    
    # Bind the source and sink to the channel
    a2.sources.r1.channels = c1
    a2.sinks.k1.channel = c1

#### 创建flume-flume2.conf ####

vim flume-flume2.conf

添加如下内容

# Name the components on this agent
    a3.sources = r1
    a3.sinks = k1
    a3.channels = c2
    
    # Describe/configure the source
    a3.sources.r1.type = avro
    a3.sources.r1.bind = hadoop102
    a3.sources.r1.port = 4142
    
    # Describe the sink
    a3.sinks.k1.type = logger
    
    # Describe the channel
    a3.channels.c2.type = memory
    a3.channels.c2.capacity = 1000
    a3.channels.c2.transactionCapacity = 100
    
    # Bind the source and sink to the channel
    a3.sources.r1.channels = c2
    a3.sinks.k1.channel = c2

### 执行配置文件 ###

分别开启对应配置文件：flume-flume2，flume-flume1，flume-netcat-flume。

bin/flume-ng agent --conf conf/ --name a3 --conf-file job/group2/flume-flume2.conf -Dflume.root.logger=INFO,console

bin/flume-ng agent --conf conf/ --name a2 --conf-file job/group2/flume-flume1.conf -Dflume.root.logger=INFO,console

bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group2/flume-netcat-flume.conf

### telnet ###

使用telnet工具向本机的44444端口发送内容

telnet localhost 44444

## 结果 ##

![\-w1296][-w1296]

[10]: /images/20220130/b97ec7ec513e40e5b84ba376cfec6c7d.png
[5ce12370e4b020998ae00b0b.png]: /images/20220130/15e62df596d44227b2d1c428a12fa056.png
[-w1296]: /images/20220130/1139d6617d254ef299d2e4a0fff8beee.png