Flume串联的实现

1. Ctrl+c 关闭 node1 的 flume，并将安装包远程拷贝到 node3

[root@node1 ~]# cd /opt/ 
[root@node1 opt]# ls 
apps data flume hadoop-2.6.5 hbase-0.98 
[root@node1 opt]# scp -r flume/ node3:/opt/

2. node3 上配置环境变量

[root@node3 ~]# vim /etc/profile 
export FLUME_HOME=/opt/flume 
export PATH=$PATH:$JAVA_HOME/bin:$ZOOKEEPER_HOME/bin:$HADOOP_HOME/bin:$H ADOOP_HOME/sbin:$HIVE_HOME/bin:$HBASE_HOME/bin:$FLUME_HOME/bin 
[root@node3 ~]# source /etc/profile

3. 将 node1 上配置文件 option 拷贝一份 option1，并修改：
   参考参数： http://flume.apache.org/releases/content/1.6.0/FlumeUserGuide.html#avro-sink

# example.conf: A single-node Flume configuration 
# Name the components on this agent 
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source 
a1.sources.r1.type = netcat
a1.sources.r1.bind = node1
a1.sources.r1.port = 44444
# Describe the sink 
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = node3
a1.sinks.k1.port = 10086
# 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

4. 将 node1 上的 option 拷贝到 node3 上，并修改 参考官网： http://flume.apache.org/releases/content/1.6.0/FlumeUserGuide.html#avro-source

node1 上操作

[root@node1 ~]# scp option node3:/root 100% 541 0.5KB/s 00:00

node3 上操作

# example.conf: A single-node Flume configuration 
# Name the components on this agent 
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source 
a1.sources.r1.type = avro
a1.sources.r1.bind = node3
a1.sources.r1.port = 10086
# Describe the sink 
a1.sinks.k1.type = logger
# 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

5. 启动 node3 上的 flume(由于 node1 发送给 node3 所以先启动 node3)

[root@node3 ~]# flume-ng agent -n a1 --conf-file option3 -Dflume.root.logger=INFO,console

再复制一个 xshell 终端：

[root@node3 ~]# ss -nal
State       Recv-Q Send-Q                                                     Local Address:Port                                                       Peer Address:Port 
LISTEN      0      128                                                                   :::22                                                                   :::*     
LISTEN      0      128                                                                    *:22                                                                    *:*     
LISTEN      0      100                                                                  ::1:25                                                                   :::*     
LISTEN      0      100                                                            127.0.0.1:25                                                                    *:*     
LISTEN      0      50                                                ::ffff:192.168.134.103:10086  #端口监听成功

6. 启动 node1 上的 flume

[root@node1 ~]# flume-ng agent -n a1 --conf-file option1 -Dflume.root.logger.INFO,console

7. Node2 上发消息给 node1

[root@node2 ~]# telnet node1 44444
Trying 192.168.134.101...
Connected to node1.
Escape character is '^]'.
HELLO YQQ
OK
HELLO NEUSOFT
OK

8. Node1 上没有显示信息，而是 node3 上显示信息，这是因为 node1 接收到消息后， 传递给了 node3

21/11/20 12:32:42 INFO ipc.NettyServer: [id: 0xb2228da4, /192.168.134.101:44199 => /192.168.134.103:10086] OPEN
21/11/20 12:32:42 INFO ipc.NettyServer: [id: 0xb2228da4, /192.168.134.101:44199 => /192.168.134.103:10086] BOUND: /192.168.134.103:10086
21/11/20 12:32:42 INFO ipc.NettyServer: [id: 0xb2228da4, /192.168.134.101:44199 => /192.168.134.103:10086] CONNECTED: /192.168.134.101:44199
21/11/20 12:35:40 INFO sink.LoggerSink: Event: { headers:{} body: 48 45 4C 4C 4F 20 59 51 51 0D                   HELLO YQQ. }
21/11/20 12:44:06 INFO sink.LoggerSink: Event: { headers:{} body: 48 45 4C 4C 4F 20 4E 45 55 53 4F 46 54 0D       HELLO NEUSOFT. }

9. 为什么需要串联呢？
   日志收集中一个非常常见的场景是大量的日志生成客户端向连接到存储子系统 的几个消费代理发送数据。 例如，从数百个 Web 服务器收集的日志发送给十几 个写入 HDFS 集群的代理
   
   Flume 支持将事件流复用到一个或多个目的地。 这是通过定义一个可以复制或 选择性地将事件路由到一个或多个通道的流复用器来实现的