Flume – 0.9.4 解析

现在自己手上维护的一个项目是基于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连接没连上!

作者: inter12

在这苦短的人生中,追求点自己的简单快乐

发表评论

电子邮件地址不会被公开。 必填项已用*标注