Flume-聚合案例

缺乏、安全感 2022-10-04 13:59 185阅读 0赞

### 聚合 ###

案例分析：

hadoop114上的Flume1监控文件/home/bd/tmp/group4Test/log

hadoop115上的Flume2监控44444端口数据流

Flume1和Flume2将数据发送给hadoop113上的Flume3，Flume3最终将数据打印到控制台。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70]

Flume1的配置文件如下

# Name the components on this agent
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    # Describe/configure the source
    a1.sources.r1.type = TAILDIR
    a1.sources.r1.filegroups = f1
    a1.sources.r1.filegroups.f1 = /home/bd/tmp/group4Test/log
    a1.sources.r1.positionFile = /home/bd/tmp/group4Position/position.json
    
    # Describe the sink
    a1.sinks.k1.type = avro
    a1.sinks.k1.hostname = hadoop113
    a1.sinks.k1.port = 4444
    
    # Use a channel which buffers events in memory
    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.sinks.k1.channel = c1

Flume2的配置文件如下

# Name the components on this agent
    a2.sources = r1
    a2.sinks = k1
    a2.channels = c1
    
    # Describe/configure the source
    a2.sources.r1.type = netcat
    a2.sources.r1.bind = localhost
    a2.sources.r1.port = 44444
    
    # Describe the sink
    a2.sinks.k1.type = avro
    a2.sinks.k1.hostname = hadoop113
    a2.sinks.k1.port = 4444
    
    # Use a channel which buffers events in memory
    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

Flume3的配置文件如下

# Name the components on this agent
    a3.sources = r2
    a3.sinks = k2
    a3.channels = c2
    
    # Describe/configure the source
    a3.sources.r2.type = avro
    a3.sources.r2.bind = hadoop113
    a3.sources.r2.port = 4444
    
    # Sink
    a3.sinks.k2.type = logger
    
    # Use a channel which buffers events in memory
    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.r2.channels = c2
    a3.sinks.k2.channel = c2

先运行a3，再运行a1和a2，再a1监控的文件追加hello和word，在a2监听的端口发送wo、shouhu、word查看结果：

![在这里插入图片描述][20210715171147366.png]![在这里插入图片描述][20210715171141676.png]

以上配置文件是两个Flume任务的avro往一个端口发送的，那么可以配置往各自端口发送，结果是一样的，配置修改如下：

Flume1

# Describe the sink
    a1.sinks.k1.type = avro
    a1.sinks.k1.hostname = hadoop113
    a1.sinks.k1.port = 4441

Flume2

# Describe the sink
    a1.sinks.k1.type = avro
    a1.sinks.k1.hostname = hadoop113
    a1.sinks.k1.port = 4442

Flume3

# Name the components on this agent
    a3.sources = r1 r2
    a3.sinks = k2
    a3.channels = c2
    
    # Describe/configure the source
    a3.sources.r1.type = avro
    a3.sources.r1.bind = hadoop113
    a3.sources.r1.port = 4441
    a3.sources.r2.type = avro
    a3.sources.r2.bind = hadoop113
    a3.sources.r2.port = 4442
    
    # Sink
    a3.sinks.k2.type = logger
    
    # Use a channel which buffers events in memory
    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.sources.r2.channels = c2
    a3.sinks.k2.channel = c2

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NsZWFybHhq_size_16_color_FFFFFF_t_70]: /images/20221004/b3cd7183e09f4d5db0f0ea611418f8c2.png
[20210715171147366.png]: https://img-blog.csdnimg.cn/20210715171147366.png
[20210715171141676.png]: /images/20221004/fc69d5ad7d60480d81ace2630acf7eaa.png