flume 简介安装使用案例(将log4j数据写到hdfs中)

chengjianxiaoxue

浏览: 1310088 次
性别:
来自: 北京

最近访客更多访客>>

liu_shui8

happy2012

nddht

yhtppp

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

flume

0 flume地址：
0) 官网地址： http://flume.apache.org/
1)官网学习：
http://flume.apache.org/documentation.html 点击 User Guide ---> 进入如下链接http://flume.apache.org/FlumeUserGuide.html
2) 下载地址：
http://flume.apache.org/download.html

1 flume简介：

cloudera公司开发的实时日志收集系统，原名是flume og (original generation),

但随着 FLume 功能的扩展，Flume OG 代码工程臃肿、核心组件设计不合理、核心配置不标准等缺点暴露出来,尤其是在 Flume OG 的最后一个发行版本 0.94.0 中，日志传输不稳定的现象尤为严重，为了解决这些问题，2011 年 10 月 22 号，cloudera 完成了 Flume-728，对 Flume 进行了里程碑式的改动：重构核心组件、核心配置以及代码架构，重构后的版本统称为 Flume NG（next generation,改动的另一原因是将 Flume 纳入 apache 旗下，cloudera Flume 改名为 Apache Flume。

这就是flume og 和 flume ng别名的取名和两者区别。

简要提下 Flume NG （1.x.x）的主要变化：

sources和sinks 使用channels 进行链接
两个主要channel 。1， in-memory channel 非持久性支持，速度快。缺点:any events still left in the memory channel when an agent process dies can’t be recovered
2 ， JDBC-based channel 持久性支持。
不再区分逻辑和物理node，所有物理节点统称为 “agents”,每个agents 都能运行0个或多个sources 和sinks
不再需要master节点和对zookeeper的依赖，以及去掉了collector，配置文件简单化。
插件化，一部分面对用户，工具或系统开发人员。
使用Thrift、Avro Flume sources 可以从flume0.9.4 发送 events 到flume 1.x

如下是flume og架构图:

具体 flume og到 flume ng到底转变了些什么，见如下链接(ibm developerworks)

http://www.ibm.com/developerworks/cn/data/library/bd-1404flumerevolution/index.html 写的不错

flume作用:

a) flume是一个分布式的数据收集系统，具有高可靠、高可用、事务管理、失败重启等功能。数据处理速度快，完全可以用于生产环境。
b) flume是分布式的日志收集系统(这里的日志是一个泛泛统称，可以是日志可以是命令行输出可以是数据文件)，把收集来的数据传送到目的地去。

2 flume架构和组成 :

a) client: 生产数据的地方

b) event: 生产的数据

c) agent: flume核心组件,接收生产的数据，暂时存储并在发送数据到目的地后删除存储数据的处理单位，

一个agent就是一个jvm, agent又是由 source, channel,sink,Interceptor等构建而成。

对上述名词具体解释如下:

Client：生产数据，运行在一个独立的线程。

Events：可以是日志记录、 avro 对象等，如果是文本文件通常是一行记录，这也是事务的基本单位

Agent：使用JVM 运行Flume。每台机器运行一个agent，但是可以在一个agent中包含多个sources和sinks。

Source：从Client收集数据，传递给Channel。不同的 source，可以接受不同的数据格式，

比如监视外部源--目录池(spooling directory)数据源，可以监控指定文件夹中的新文件变化，

如果目录中有文件产生，就会立刻读取其内容。

source组件可以处理各种格式的日志数据,eg:avro Sources、thrift Sources、exec、jms、spooling directory、netcat、sequence generator、syslog、http、legacy、自定义。
支持的这些格式都可以通过http://flume.apache.org/documentation.html 点击 User Guide ---> http://flume.apache.org/FlumeUserGuide.html查询到

Channel：连接 sources 和 sinks ，这个有点像一个队列，是一个存储池，接收source的输出，直到有sink

消费掉channel中的数据，channel中的数据直到进入下一个channel或者进入sink才会被删除，

当sink写入失败后，可以自动重启，不会造成数据丢失。

临时存放的数据可以存放在memory Channel、jdbc Channel、file Channel、自定义。

Sink：从Channel收集数据，运行在一个独立线程。用于把数据发送到目的地的组件

目的地包括hdfs、logger、avro、thrift、ipc、file、null、hbase、solr、自定义。

在整个数据传输过程中，流动的是event。事务保证是在event级别
flume可以支持多级flume的agent，(即多个flume可以连成串，上一个flume可以把数据写到下一个flume上)
支持扇入(fan-in)、扇出(fan-out)
扇入(fan-in): source可以接受多个输入，
扇出(fan-out): sink可以输出到多个目的地

flume ng节点组成图:

多Agent并联下的架构图(flume的web架构)：

3 flume特性:

a) 可靠性

当节点出现故障时，日志能够被传送到其他节点上而不会丢失。

Flume提供了三种级别的可靠性保障，从强到弱依次分别为：

end-to- end（收到数据agent首先将event写到磁盘上，当数据传送成功后，再删除；如果数据发送失败，可以重新发送。）

Store on failure（这也是scribe采用的策略，当数据接收方crash崩溃时，将数据写到本地，待恢复后，继续发送）

Best effort（数据发送到接收方后，不会进行确认）

b) 可扩展性

物理可扩展性：

Flume采用了三层架构，分别为agent，collector(clint端吧)和storage(sink数据流向方吧)，每一层均可以水平扩展。其中，所有agent和 collector由master统一管理，这使得系统容易监控和维护，且master允许有多个（使用ZooKeeper进行管理和负载均衡），这就避免了单点故障问题。(这是针对flume og在结合zk情况下)

逻辑可扩展性：

用户可以根据需要添加自己的agent，collector或者storage。此外，Flume自带了很多组件，包括各种agent（file，syslog等），collector和storage（file，HDFS等），就是flume可以采集多种类型的数据，数据最后可以流向保存到多种类型的目的中。

c) 可管理性 这是针对flume og在结合zk情况下

所有agent和Collector由master统一管理，这使得系统便于维护。多master情况，Flume利用 ZooKeeper和gossip，保证动态配置数据的一致性。用户可以在master上查看各个数据源或者数据流执行情况，且可以对各个数据源配置和动态加载。Flume提供了web 和shell script command两种形式对数据流进行管理。

4 flume安装：

解压：
[root@h2sliver112 conf]# tar -zxvf apache-flume-1.5.2-bin.tar.gz 
重命名：
[root@h2sliver112 local]# mv apache-flume-1.5.2-bin flume1.5.2-bin
增加配置文件
[root@h2sliver112 flume1.5.2-bin]# cd conf
[root@h2sliver112 conf]# cp flume-env.sh.template flume-env.sh
[root@h2sliver112 conf]# cp flume-conf.properties.template flume-conf.properties
[root@h2sliver112 conf]# vi flume-env.sh 修改JAVA_HOME=/opt/jdk1.7
[root@h2sliver112 conf]# 验证 flume-ng  version
Flume 1.5.2
Source code repository: https://git-wip-us.apache.org/repos/asf/flume.git
Revision: 229442aa6835ee0faa17e3034bcab42754c460f5
Compiled by hshreedharan on Wed Nov 12 12:51:22 PST 2014
From source with checksum 837f81bd1e304a65fcaf8e5f692b3f18

5 flume简单案例：使用avro方式监听一个文件并将文件通过log4j样式输出到console

0 目的：写一个简单案例(通过avro方式，从客户端上读取一个文件，然后提交到avro服务端的source获取，通过内存channel最后将数据输送到目的地logger 并在控制台输出)
testflume----> agent-source-avro----> agent-channel-memory--->agent-sink-flume-logger--->-Dflume.root.logger=DEBUG,console控制台打印 

1 对应agent配置文件写法为： 
这里我将文件写在目录：
# pwd
/usr/local/flume1.5.2-bin/conf内

[root@h2sliver112 conf]# vi agent1.conf

内容如下：
agent1.sources=source1
agent1.channels=channel1
agent1.sinks=sink1


agent1.sources.source1.type=avro   表示读取数据源格式为avro
agent1.sources.source1.bind=0.0.0.0
agent1.sources.source1.port=41414   端口表示这个client输送到这个agent要走的socket端口， 定义的agent不同端口下表示接受不同client发送来的数据。
agent1.sources.source1.channels=channel1

agent1.channels.channel1.type=memory

agent1.sinks.sink1.type=logger  表示agent导出数据打印到flume日志文件中
agent1.sinks.sink1.channel=channel1


2 根据上面配置好的agent.conf配置信息，启动flume agent。
[root@h2sliver112 bin]# flume-ng agent  --conf ../conf/ -f ../conf/agent1.conf -n agent1 -Dflume.root.logger=DEBUG,console

--conf ../conf/ 指定flume配置文件目录位置
-f指定flume服务端启动监听的agent配置文件  

-n指定启动的flume agent别名

-Dflume.root.logger=DEBUG,console 表示数据在控制台打印出

启动flume agent后，可以看到如下信息：
 Avro source source1 started，此时会不停扫描agent1.conf

上面配置sink为flume logflume logger,配置内容为：
more conf/log4j.properties:

#for test can use  -Dflume.root.logger=DEBUG,console when launching flume 注释表示如果是测试那么可以用这种方式

flume.root.logger=INFO,LOGFILE 默认是将数据写入到当前目录下logs文件夹内的文件名为flume.log的文件中

flume.log.dir=./logs

flume.log.file=flume.log



3 本地文件如下：
[root@h2sliver112 local]# cat testflume 
1
2
3
4
5

3 启动avro client，读取本地文件，在看上面启动的服务端是否有数据输出：
启动avro client命令如下：
[root@h2sliver112 bin]# flume-ng avro-client --conf ../conf/ -H localhost -p 41414 -F /usr/local/testflume 
-H localhost 指定运行机器
-p 41414 指定端口
-F /usr/local/testflume 指定外部数据源文件
上述命令启动后，flume avro client将文件testflume数据输送到上面定义的 flume agent source中，

avro-client           run an avro Flume client
avro-client 
options:
  --host,-H <host>       hostname to which events will be sent
  --filename,
-F <file>   text file to stream to avro source
 
可以看到监听到的服务端打印结果如下：见截图，可见将文件数据监听的每个Event都打印出来了。

上述案例中，如果启动 flume agent为:

flume-ng agent --conf ../conf/ -f ../conf/agent1.conf -n agent1 ，那么最后产生数据会在：
[root@hadoop3 logs]# pwd
/opt/flume1.5.2/bin/logs
[root@hadoop3 logs]# ll
total 12
-rw-r--r-- 1 root root 11215 Feb 21 00:56 flume.log

如果启动flume agent想启动后在后台执行，那么写成:

flume-ng agent --conf ../conf/ -f ../conf/agent1.conf -n agent1& 加上&即可。

5 flume使用案例2，使用 avro将log4j不断产生的日志数据写到hdfs中：

需要准备的：

a) hadoop2环境，并事先创建好hdfs目录： /flume/events

b) 包flume-ng-log4jappender-1.5.0-cdh5.1.3-jar-with-dependencies.jar 用于将log4j的信息和flume关联,附件可下载此工程和对应包

开工：

0 如果使用maven 需要下载 flume slf4j的依赖：

<!-- flume -->
		<dependency>
			<groupId>org.apache.flume</groupId>
			<artifactId>flume-ng-core</artifactId>
			<version>1.5.2</version>
			<exclusions>
				<exclusion>
					<groupId>org.slf4j</groupId>
					<artifactId>slf4j-api</artifactId>
				</exclusion>
				<exclusion>
					<groupId>org.slf4j</groupId>
					<artifactId>slf4j-log4j12</artifactId>
				</exclusion>
				<exclusion>
					<groupId>log4j</groupId>
					<artifactId>log4j</artifactId>
				</exclusion>
			</exclusions>
		</dependency>
 <dependency>
      <groupId>org.apache.flume.flume-ng-clients</groupId>
      <artifactId>flume-ng-log4jappender</artifactId>
      <version>1.5.2</version>
    </dependency>

1 Java端:

1 java

不断产生日志，模拟web系统不停运行产生日志效果
public class GenerateLog4j {

	/**
	 * @param args
	 * @throws InterruptedException 
	 */
	public static void main(String[] args) throws InterruptedException {
		// TODO Auto-generated method stub
		while(true){
			Logger logger = org.apache.log4j.Logger.getLogger(GenerateLog4j.class);
			logger.error("日期时间" + System.currentTimeMillis());
			Thread.sleep(1000);
		}
		 
	}


2 Java端的log4j.properties

log4j.rootLogger=INFO,flume
log4j.appender.flume = org.apache.flume.clients.log4jappender.Log4jAppender
log4j.appender.flume.Hostname = 192.168.1.112 // 对应flume 安装的位置机器地址
log4j.appender.flume.Port = 44444  // 对应flume启动的监听端口 这个端口在flume 定义的conf中会写好
log4j.appender.flume.UnsafeMode = true


3 flume:
flume conf/创建 agent2.conf 内容如下
agent2.sources=source1
agent2.channels=channel1
agent2.sinks=sink1

agent2.sources.source1.type=avro
agent2.sources.source1.bind=0.0.0.0
agent2.sources.source1.port=44444    // 指定监听端口
agent2.sources.source1.channels=channel1

agent2.sources.source1.interceptors = i1 i2   指定使用flume的两个拦截器，一个是时间的 一个是IP的
agent2.sources.source1.interceptors.i1.type = org.apache.flume.interceptor.HostInterceptor$Builder
agent2.sources.source1.interceptors.i1.preserveExisting = true
agent2.sources.source1.interceptors.i1.useIP = true
agent2.sources.source1.interceptors.i2.type = org.apache.flume.interceptor.TimestampInterceptor$Builder

agent2.channels.channel1.type=memory
agent2.channels.channel1.capacity=10000
agent2.channels.channel1.transactionCapacity=1000
agent2.channels.channel1.keep-alive=30

agent2.sinks.sink1.type=hdfs
agent2.sinks.sink1.channel=channel1
agent2.sinks.sink1.hdfs.path=hdfs://h2single:9000/flume/events/%{host}/%y-%m-%d   指定写到目标hadoop2集群的hdfs某个目录下 可以用hive创建分区表加载并继续做MR操作
agent2.sinks.sink1.hdfs.fileType=DataStream
agent2.sinks.sink1.hdfs.writeFormat=Text
agent2.sinks.sink1.hdfs.rollInterval=0
agent2.sinks.sink1.hdfs.rollSize=10000
agent2.sinks.sink1.hdfs.rollCount=0
agent2.sinks.sink1.hdfs.idleTimeout=5


4 flume目录下启动如下命令：
bin/flume-ng agent --conf ./conf/  -Dflume.monitoring.type=http -Dflume.monitoring.port=34343 -n agent2 -f conf/agent2.conf
// flume应用参数监控 -Dflume.monitoring.port=34343 可以通过http://ip:34343/metrics访问，从而看到监控信息
5 运行Java端GenerateLog4j.java

6 查看h2single hdfs被写入的数据  数据截图如下：