Hadoop数据收集系统—Flume
Flume OG
OG：“Original Generation”
0.9.x或cdh3以及更早版本
由agent、collector、master等组件构成
Flume NG
NG：“Next/New Generation”
1.x或cdh4以及之后的版本
由Agent、Client等组件构成
为什么要推出NG版本
精简代码
架构简化
Flume OG基本架构
Agent
用于采集数据
数据流产生的地方
通常由source和sink两部分组成
Source用于获取数据，可从文本文件，syslog，HTTP等获取数据；
Sink将Source获得的数据进一步传输给后面的Collector。
Flume自带了很多source和sink实现
syslogTcp(5440) | agentSink("localhost",35856)
tail("/etc/service_files") | agentSink("localhost",35856)
Collector
汇总多个Agent结果
将汇总结果导入后端存储系统，比如HDFS，HBase
Flume自带了很多collector实现
collectorSource(35856) | console
CollectorSource(35856) | collectorSink("file:///tmp/flume/collected", "syslog");
collectorSource(35856) | collectorSink("hdfs://namenode/user/flume/ ","syslog");
Agent与Collector对应关系
Agent与Collector对应关系
可手动指定，也可自动匹配
自动匹配的情况下，master会平衡collector之间的负载。
问题：为什么引入Collector？
对Agent数据进行汇总，避免产生过多小文件；
避免多个agent连接对Hadoop造成过大压力 ；
中间件，屏蔽agent和hadoop间的异构性。
Master
管理协调 agent 和collector的配置信息；
Flume集群的控制器；
跟踪数据流的最后确认信息，并通知agent；
通常需配置多个master以防止单点故障；
借助zookeeper管理管理多Master。
容错机制
三种可靠性级别
agentE2ESink[("machine"[,port])]
gent收到确认消息才认为数据发送成功，否则重试.
agentDFOSink[("machine"[,port])]
当agent发现在collector操作失败的时候，agent写入到本地硬盘上，当collctor恢复后，再重新发送数据。
agentBESink[("machine"[,port])]
效率最好，agent不写入到本地任何数据，如果在collector 发现处理失败，直接删除消息。
构建基于Flume的数据收集系统
1. Agent和Collector均可以动态配置
2. 可通过命令行或Web界面配置
3. 命令行配置
在已经启动的master节点上，依次输入”flume shell”è”connect localhost ”
如执行 exec config a1 ‘tailDir(“/data/logfile”)’ ‘agentSink’
4. Web界面
选中节点，填写source、sink等信息
常用架构举例—拓扑1
agentA : tail(“/ngnix/logs”) | agentSink("collector",35856);
agentB : tail(“/ngnix/logs”) | agentSink("collector",35856);
agentC : tail(“/ngnix/logs”) | agentSink("collector",35856);
agentD : tail(“/ngnix/logs”) | agentSink("collector",35856);
agentE : tail(“/ngnix/logs”) | agentSink("collector",35856);
agentF : tail(“/ngnix/logs”) | agentSink("collector",35856);
collector : collectorSource(35856) | collectorSink("hdfs://namenode/flume/","srcdata");
常用架构举例—拓扑2
agentA : src | agentE2ESink("collectorA",35856);
agentB : src | agentE2ESink("collectorA",35856);
agentC : src | agentE2ESink("collectorB",35856);
agentD : src | agentE2ESink("collectorB",35856);
agentE : src | agentE2ESink("collectorC",35856);
agentF : src | agentE2ESink("collectorC",35856);
collectorA : collectorSource(35856) | collectorSink("hdfs://...","src");
collectorB : collectorSource(35856) | collectorSink("hdfs://...","src");
collectorC : collectorSource(35856) | collectorSink("hdfs://...","src");
常用架构举例—拓扑3
agentA : src | agentE2EChain("collectorA:35856","collectorB:35856");
agentB : src | agentE2EChain("collectorA:35856","collectorC:35856");
agentC : src | agentE2EChain("collectorB:35856","collectorA:35853");
agentD : src | agentE2EChain("collectorB:35853","collectorC:35853");
agentE : src | agentE2EChain("collectorC:35853","collectorA:35853");
agentF : src | agentE2EChain("collectorC:35853","collectorB:35853");
collectorA : collectorSource(35853) | collectorSink("hdfs://...","src");
collectorB : collectorSource(35853) | collectorSink("hdfs://...","src");
collectorC : collectorSource(35853) | collectorSink("hdfs://...","src");
