协作框架之Flume
1.概念
Flume
Cloudera 公司开源的框架
高效的收集海量日志文件
官网
应用场合
日志来源于apache/Nginx 应用服务器的日志 --> HDFS
Flume+kafka ---> Storm /Spark Streaming
2.简单架构（配置）
Agent --> 每台应用服务器日志的机器 运行一个agent
source/channel/sink
source 应用服务器的日志目录或文件，数据源
channel source主动将数据pash到管道中，内存和磁盘
安全性，可靠性，借助管道写道本地磁盘
sink： 写到HDFS
Event是数据传输的基本单元
Flume以事件的形式将数据从源头传送到最终目的地
Event由可选的header和载有数据的一个byte array构成
载有的数据flume是不透明的
Header是容纳了key-value字符串对的无序集合，key在集合内是唯一的。
Header可以在上下文路由中使用扩展
企业中应用时，日志按日期生成，可以将服务器的时间放到头里，数据真正放到byte array中
3. Flume的特点
3.1复杂流动性
Flume允许用户进行多级流动到最终目的地，也允许扇出流（一到多）、扇入流(多到一)的、故障转移和失败处理。fan-in fan-out
3.2可靠性
事务性的数据传递，保证了数据的可靠性。
3.3 可恢复性
通道可以以内存或文件的方式实现，内存更快，但是不可恢复，而文件虽然比较慢但提供了可恢复性。
*运行在有日志的地方
*系统：linux
*JVM/JDK
*轻量级的服务（e.g.：zk jn zkfc sqoop对硬件要求不是很高）
4.Flume的安装和部署
下载上传 cdh版本
1).解压
$ tar -zxf flume-ng-1.5.0-cdh5.3.6.tar.gz -C /opt/modules/cdh/
2).配置${FLUME_HOME}/conf/
$ cp flume-env.sh.template flume-env.sh
修改flume-env.sh
3).拷贝HDFS依赖及配置的jar包到${FLUME_HOME}/lib
${HADOOP_HOME}/
$ cp share/hadoop/common/hadoop-common-2.5.0-cdh5.3.6.jar share/hadoop/common/lib/commons-configuration-1.6.jar share/hadoop/common/lib/hadoop-auth-2.5.0-cdh5.3.6.jar share/hadoop/hdfs/hadoop-hdfs-2.5.0-cdh5.3.6.jar /opt/modules/cdh/flume-1.5.0-cdh5.3.6/lib/
4).拷贝HDFS相关的配置文件到flume的conf
${HADOOP_HOME}/etc/hadoop --> ${FLUME_HOME}/conf
$ cp etc/hadoop/core-site.xml etc/hadoop/hdfs-site.xml /opt/modules/cdh/flume-1.5.0-cdh5.3.6/conf/
[案例一： source：telnet sink:生成日志文件，直接打印到控制台]
1.生成agent模板
$ cp flume-conf.properties.template flume-conf.properties
$ cp flume-conf.properties a1.conf
2.编辑agent配置文件（source channel sink）
# Name the components on this agent
#a1指的是Agent的name，需要与启动agent的--name相对应
a1.sources = r1 #当前agent的sources名称
a1.sinks = k1 #当前agent的sinks名称
a1.channels = c1 #当前agent的channels名称
# Describe/configure the source
a1.sources.r1.type = netcat #sources的类型
a1.sources.r1.bind = bigdata.ibeifeng.com #绑定的主机
a1.sources.r1.port = 44444 #监听的端口
# 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 #存储到channels中的events的最大数据
a1.channels.c1.transactionCapacity = 100 数量##每次event在chnnel传输的最大的
# Bind the source and sink to the channel 将对应的source和sink绑定到channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
3.yum 安装 telnet
# yum -y install telnet
4.启动agent ${FLUME_HOME}/
$ bin/flume-ng agent --conf conf --conf-file conf/a1.conf --name a1 -Dflume.root.logger=INFO,console
--conf 指定配置文件所在目录
--name 指定agent的名称,与a1.conf文件指定的一致
--conf-file 指定agent配置文件名称
-Dflume.root.logger=INFO,console 日志输出到console
Event: { headers:{} body: 48 65 6C 6C 6F 20 77 6F 72 6C 64 21 0D Hello world!. }
5.启动telnet
$ telnet bigdata.ibeifeng.com 44444
【退出telnet】
Ctrl+]
telnet>quit
Connection closed.
6.停止agent
${FLUME_HOME}
Ctrl+c 退出flume agent同时也就退出了telnet
[案例二：flume抽取日志文件]
source: 类型exec
tail -f
channel:memchannel
sink:HDFS
//使用agent a1 作为模板生成a2 agent 的配置文件
${FLUME_HOME}/conf
$ cp a1.conf a2.conf
1.配置a2.conf
=================修改a2.conf
#a2:agent name
a2.sources = r2
a2.channels = c2
a2.sinks = k2
# define sources
#主动获取日志
a2.sources.r2.type = exec
#获取日志的命令（注意要有权限）
a2.sources.r2.command = tail -F /var/log/httpd/access_log
#上一行命令所运行的环境
a2.sources.r2.shell = /bin/bash -c
# define channels
a2.channels.c2.type = memory
a2.channels.c2.capacity = 1000
a2.channels.c2.transactionCapacity = 100
# define sinks
#目标上传到hdfs
a2.sinks.k2.type = hdfs
a2.sinks.k2.hdfs.path=hdfs://[hostname]:8020/flume/%Y%m%d/%H
a2.sinks.k2.hdfs.filePrefix = accesslog
#启用按时间生成文件夹
a2.sinks.k2.hdfs.round=true
#设置roundValue:1，round单位：小时
a2.sinks.k2.hdfs.roundValue=1
a2.sinks.k2.hdfs.roundUnit=hour
#使用本地时间戳（这个必须设置不然会报错）
a2.sinks.k2.hdfs.useLocalTimeStamp=true
#多少个events会flush to hdfs
a2.sinks.k2.hdfs.batchSize=1000
# File format: 默认是SequenceFile（key:value对），DataStream是无压缩的一般数据流
a2.sinks.k2.hdfs.fileType=DataStream
#序列化的格式Text
a2.sinks.k2.hdfs.writeFormat=Text
#设置解决文件过多、过小问题
#每600秒生成一个文件
a2.sinks.k2.hdfs.rollInterval=60
#当达到128000000bytes时，创建新文件 127*1024*1024（in bytes）
#实际环境中如果按照128M回滚文件,那么这里设置一般设置成127M
a2.sinks.k2.hdfs.rollSize=128000000
#设置文件的生成不和events数相关
a2.sinks.k2.hdfs.rollCount=0
#设置成1，否则当有副本复制时就重新生成文件，上面三条则没有效果
a2.sinks.k2.hdfs.minBlockReplicas=1
# bind the sources and sinks to the channels
a2.sources.r2.channels = c2
a2.sinks.k2.channel = c2
===================================
2.安装Apache HTTP服务器程序用于生成网站日志文件
2.1 安装Apache HTTP
# yum -y install httpd
2.2 启动httpd服务
# service httpd start
2.3 编辑一个静态的html的页面
# vi /var/www/html/index.html
this is a test html
2.4 浏览器输入主机名访问这个页面
bigdata.ibeifeng.com
2.5 实时监控httpd日志
# chmod -R 777 /var/log/httpd
$ tail -f /var/log/httpd/access_log
3.启动hadoop
$ sbin/start-dfs.sh
4.启动Flume-agent a2
$ bin/flume-ng agent --conf conf --conf-file conf/a2.conf --name a2 -Dflume.root.logger=INFO,console
5.刷新静态页面，观察HDFS是否生成指定的目录和文件
[案例三：flume抽取目录]
source: 类型spooldir
channel:memchannel
sink:HDFS
//使用agent a2 作为模板生成a3 agent 的配置文件
${FLUME_HOME}/conf
$ cp a2.conf a3.conf
1.配置a3.conf
=================修改a3.conf
a3.sources = r3
a3.sinks = k3
a3.channels = c3
# Describe/configure the source
# 源是某个目录使用spooldir
a3.sources.r3.type = spooldir
# 抽取的目录
a3.sources.r3.spoolDir = /opt/modules/cdh/hadoop-2.5.0-cdh5.3.6/logs
# 抽取该目录下符合包含.log结尾的文件
a3.sources.r3.includePattern = ^.log$
# Use a channel which buffers events in file
# 设置channel类型是file
a3.channels.c3.type = file
# 设置检查点目录，记录已经获取哪些文件，一些元数据信息
a3.channels.c3.checkpointDir = /opt/modules/cdh/flume-1.5.0-cdh5.3.6/checkpoint
#设置缓存的数据存储目录
a3.channels.c3.dataDirs = /opt/modules/cdh/flume-1.5.0-cdh5.3.6/bufferdata
# Describe the sink
a3.sinks.k3.type = hdfs
# 启用设置多级目录，这里按年/月/日/时 2级目录，每个小时生成一个文件夹
a3.sinks.k3.hdfs.path = hdfs://bigdata.ibeifeng.com:8020/flume2/%Y%m%d/%H
# 设置HDFS生成文件的的前缀
a3.sinks.k3.hdfs.filePrefix = accesslog
#启用按时间生成文件夹
a3.sinks.k3.hdfs.round = true
#设置round单位:小时
a3.sinks.k3.hdfs.roundValue = 1
a3.sinks.k3.hdfs.roundUnit = hour
#使用本地时间戳
a3.sinks.k3.hdfs.useLocalTimeStamp = true
# 设置每次写入的DFS的event的个数为100个
a3.sinks.k3.hdfs.batchSize = 100
# 写入HDFS的方式
a3.sinks.k3.hdfs.fileType = DataStream
# 写入HDFS的文件格式
a3.sinks.k3.hdfs.writeFormat = Text
#设置解决文件过多过小问题
#每600秒生成一个文件
a3.sinks.k3.hdfs.rollInterval = 60
#当达到128000000bytes时，创建新文件 127*1024*1024
#实际环境中如果按照128M回顾文件,那么这里设置一般设置成127M
a3.sinks.k3.hdfs.rollSize = 128000000
#设置文件的生成不和events数相关，与时间和大小相关
a3.sinks.k3.hdfs.rollCount = 0
#设置成1，否则当有副本复制时就重新生成文件，上面三条则没有效果
a3.sinks.k3.hdfs.minBlockReplicas =1
# Bind the source and sink to the channel
a3.sources.r3.channels = c3
a3.sinks.k3.channel = c3
===================================
2.启动Flume-agent a3
$ bin/flume-ng agent --conf conf --conf-file conf/a3.conf --name a3 -Dflume.root.logger=INFO,console
[扩展]
Flume
http://m.blog.csdn.net/article/details?id=51892945
网友评论