知行天下

现在自己手上维护的一个项目是基于Flume0.9.4这个版本做的数据传输，所以必须去了解下Flume这个版本的一些设计.下面提到的Flume都指0.9.4这个版本，除非特别指出是其他版本。
这个世界上，没有坏的工具，只有把工具用坏的人。能做一个简单系统的才是牛逼的人。做设计其实跟读书没什么不同，读书的过程是先把书读厚，再读薄，设计也是如此，先把系统做复杂，再把系统做简单。
这篇不会按部就班的去解构Flume的一些东西，没什么意义。会按照自己的理解过程来做剖析。

Flume是凭什么吃饭的？
Flume是业界知名的大数据公司Cloudera出品的一个工具，用来解决一些大数据量的数据搬迁工作。
简单的说就是期望把散落到各个服务器上的日志信息，通过Flume搬迁到任何一个地方，例如HDFS，HBASE，MQ等。
搬迁的方式是流式的。

Flume是怎么干的？
既然是做数据转移，应该就有一个数据源的采集和数据的转发终端，在Flume中一个核心组件就是Node，一个Node就是一个节点。
Node包含两个部分：source和sink。

• source：负责采集哪里的数据，也就是数据的输入
• sink：负责数据的转发，也就是将采集的数据转发到哪里去。

如何通信？
若是有两个node,那么必然面临一个node的sink做为另一个node的source，两者之间就存在远程通信的问题。
flume支持的远程RPC协议有：THRIFT/AVRO

有哪些输出格式？

• avro  ： avro格式，默认不压缩
• avrodata ： avro格式，编码过
• avrojson ： avro格式，json方式组织
• default  ：一种dubug模式
• json  ：json是组织
• log4j  ：log4j的日志格式
• raw  ：最为原始的信息，不包含 host/timestamp/nanos.
• syslog  ： syslog格式
• seqfile  ： 二进制的hadoop格式

如何管理？
如果有很多Node的话，怎么管理这些Node就成为一个问题，也就出现了master节点。
master:管理所有的Node，负责他们之间的心跳，配置之类信息。
master要做配置的话，那么必然需要存储一些状态信息，Flume-master支持两种方式存储：内存和zk(zk分为内置zk和外置zk)

可靠性
到这步，就可以清楚Flume这个版本用的还是集中式管理的架构，那么master就必然面临如何做到高可用。
对于高可用通用的做法就是备份–HA。flume也是这么做的。
如此Flume中最为重要的两个节点都已经出现了。

一层的Node够吗？
通过mater作为调度管理,node即作为数据的采集，数据通过远程RPC输出，这种结构带来的问题是整个网络结构的非常复杂。
Flume是通过将Node分为了 Agent和Collector两层来解决这个问题的。

• Agent：部署到业务服务器上负责收集数据，并向后端的Collector转发
• Collector：监听端口，并聚合数据到终端.

到这里它的整体架构图就如下所示：

更多的source和sink:

Node中默认的source的信息：

• collectorSource(port):开启应用指定的port，监听数据，一般对应的就是agentSink(“ip”,port);
• autoCollectorSource:自动收集数据源
• logicalSource：信息不完全而且抽象的source，由master自动管理，当master获取到足够信息时候将其转化为rpcSource

还有些基本的source
tail,text,console,rpcSource(port),syslogTcp(port),syslogUdp(port)
Node中的Collector层 sink的信息：
collectorSink:数据转发到hdfs的sink

Node中的Agent层 sink的信息：

• agentSink[(“machine”[, port])]：最为基本的sink，将数据转发到指定的地址和端口，等同于agentE2ESink
• agentE2ESink[(“machine”[, port])]:可靠性最好的sink，每条数据都会先保存到本地，有ack机制确认，WAL。
• agentDFOSink[(“machine”[, port])：若是数据传输失败的话，会存储到本地，再次发送
• agentBESink[(“machine”[, port])]：效率最高的sink,数据不保存，丢失也不处理。
• agentE2EChain(“m1[:_p1_]”[, “m2[:_p2_]”[,…]])：优先使用第一个m1：p1，若是失败的话，顺序使用下一个
• agentDFOChain(“m1[:_p1_]”[, “m2[:_p2_]”[,…]])：优先使用第一个m1：p1，若是失败的话，顺序使用下一个
• agentBEChain(“m1[:_p1_]”[, “m2[:_p2_]”[,…]])：优先使用第一个m1：p1，若是失败的话，顺序使用下一个
• autoE2EChain：同agentE2ESink，不过交由master控制
• autoDFOChain：同agentDFOSink，不过交由master控制
• autoBEChain：同agentBESink，不过交由master控制

Node中逻辑sink
logicalSink(“logicalnode”)：需要定义一个逻辑节点，并由master控制流向，很少用！

Node中基本sink
console,text,dfs,rpcsink,tail,syslog

可伸缩
所有的collector节点是可以自由增加的

Flume的权限认证
Kerberos

Flume的对手们？
在大数据的整个平台下，跟它类似或是可替代它的开源产品有以下：

• Sqoop：一个批量数据传输工具，主要解决的是关系数据库中数据结合到hadoop平台中的问题
• Amazon Kinesis:实时流式数据传输和处理服务，可用于不同源的流式数据采集、分流和整合计算,亚马逊出品
• SLS：针对大规模日志数据，进一步整合日志收集、存储、查询分析的服务，阿里云出品
• DataX:淘宝的开源数据传输工具，目前在淘宝有大量的应用
• Blackhole:大众点评数据部门出品的数据传输工具，支持流式传输

Flume-0.9.4版本存储的问题

1. 限定死了node的类型，分为了agent和collector，设计的复杂带来了可复用组合的降低
2. source和sink之间没有缓冲，在压力非常大的情况下，需要自己去写本地缓冲：内存或是磁盘
3. master中心节点，集中式的架构，虽然做了HA，在节点数量非常多的情况下，master会成为瓶颈，所有集中式管理模式的架构都存在这个问题

Flume实战：
1.下载解压
wget https://github.com/downloads/cloudera/flume/flume-distribution-0.9.4-bin.tar.gz
tar -zxvf flume-distribution-0.9.4-bin.tar.gz

2.配置文件修改

 mkdir /home/inter12/soft/flume/0.9.4 #建立目录
 mv flume-distribution-0.9.4-bin master
 cd /home/inter12/soft/flume/0.9.4/master/bin
 cp flume-env.sh.template flume-env.sh
 vim flume-env.sh #新增以下配置
 export JAVA_HOME=/usr/local/jdk
 export FLUME_HOME=/home/inter12/soft/flume/0.9.4/master
 export FLUME_CONF_DIR=$FLUME_HOME/conf
 export PATH=$PATH:$FLUME_HOME/bin

cd /home/inter12/soft/flume/0.9.4/master/conf
vim flume-size.xml
<property>
 <name>flume.master.servers</name>
 <value>localhost</value>
 <description>This is the address for the config servers status
 server (http)
 </description>
 </property>

将master修改为localhost

3.启动master
cd /home/inter12/soft/flume/0.9.4/master/bin
./flume master

打开http://localhost:35871
master节点搭建完毕。

再建立agent节点
cp -r /home/inter12/soft/flume/0.9.4/master agent
cd /home/inter12/soft/flume/0.9.4/agent/bin
./flume node_nowatch -n node1

建立collector节点
cp -r /home/inter12/soft/flume/0.9.4/master collector
cd /home/inter12/soft/flume/0.9.4/agent/conf
sed -i ‘s/35862/35863/g’ flume-conf.xml
cd /home/inter12/soft/flume/0.9.4/collector/bin
./flume node_nowatch -n node2

配置管理平台
打开 http://localhost:35871/flumeconfig.jsp
在Configure multiple nodes 中输入
node1:tail(“/tmp/11”)|agentSink(“localhost”,35853);
node2:collectorSource(35853)|text(“/tmp/11.copy”);

从/tmp/11中读取文件，并发送到/tmp/11.copy文件

若是出现一下情况：
inter12@inter12:~/soft/flume/0.9.4/agent$ netstat -anpl | grep 35853
(Not all processes could be identified, non-owned process info
will not be shown, you would have to be root to see it all.)
tcp6 0 0 :::35853 :::* LISTEN 9273/java
tcp6 0 0 127.0.0.1:46282 127.0.0.1:35853 TIME_WAIT –

那么重启下agent，这个是因为collector在35853端口监听，但是agent连接没连上！