一.Apache Flume

1.概述

Flume是Cloudera提供的一个高可用的，高可靠的，分布式的海量日志采集聚合和传输的软件。

Flume的核心是把数据从数据源(source)收集过来，再将收集到的数据送到指定的目的地(sink)，为了保证传输过程一定成功，在送到目的地(sink)之前，会先缓存(channel)，待数据真正到达目的地(sink)之后，Flume再删除自己缓存的数据。

Flume支持定制各类数据发送方，用于收集各类型数据；同时，Flume支持定制各种数据接受方，用于最终存储数据。一般的采集需求，通过对Flume的简单配置即可实现。针对特殊场景也具备良好的自定义扩展能力。因此，Flume可以适应大部分的日常数据采集情景。

当前Flume有两个版本，Flume 0.9X版本的统称为Flume OG，而1.X版本统称为Flume NG，后来Flume纳入Apache旗下，Cloudera Flume改名为Apache Flume。

2.运行机制

Flume系统中核心的角色是agent，agent本身是一个Java进程，一般运行在日志收集节点

每一个agent相当于一个数据传输员 ，内部有三个组件：

source：采集源，用于和数据源对接，以获取数据，Source是数据的收集端，负责将数据捕获后进行特殊的格式化，将数据封装到事件（event） 里，然后将事件推入Channel中。Flume提供了很多内置的Source， 支持 Avro， log4j， syslog 和 http post(body为json格式)。可以让应用程序同已有的Source直接打交道，如AvroSource，SyslogTcpSource。 如果内置的Source无法满足需要， Flume还支持自定义Source。；

sink：Sink从Channel中取出事件，然后将数据发到别处，可以向文件系统、数据库、 hadoop存数据， 也可以是其他agent的Source。在日志数据较少时，可以将数据存储在文件系统中，并且设定一定的时间间隔保存数据。

channel：Channel是连接Source和Sink的组件，大家可以将它看做一个数据的缓冲区（数据队列），它可以将事件暂存到内存中也可以持久化到本地磁盘上，直到Sink处理完该事件。介绍两个较为常用的Channel， MemoryChannel和FileChannel。；

在整个数据的传输的过程中，流动的是event，它是Flume内部传输的基本单位，它将传输的数据进行封装，如果是文本文件，通常是一行记录，本身为一个字节数组，并且可以携带headers(头信息)。event代表着一个数据的最小完整单元，

向外部数据源来，向外部目的地去.

3.Flume采集系统结构图

1)简单结构

2)复杂结构

二.安装部署

Flume的安装是Hadoop生态圈中最简单的——上传安装包并解压后，我们只需修改flume-env.sh，配置JAVA_HOME即可，Flume不一定非要安装在数据源节点上，只要Flume所在节点与数据源所在节点能以某种方式通信就好啦。

接下来自定义一个采集方案测试Flume——vi netcat-logger.conf：

#agent中各组件的名字
##表示agent中的source组件
a1.sources = r1
##表示的是下沉组件sink
a1.sinks = k1
##agent内部的数据传输通道channel，用于从source将数据传递到sink
a1.channels = c1

#描述和配置source组件：r1
##netcat用于监听一个端口的
a1.sources.r1.type = netcat
##配置的绑定地址,这个机器的hostname是node1,所以下面也可以配置成node1
a1.sources.r1.bind = node1
##配置的绑定端口
a1.sources.r1.port = 44444

#描述和配置sink组件：k1
a1.sinks.k1.type = logger

##描述和配置channel组件，此处使用时内存缓存的方式
#下面表示的是缓存到内存中，如果是文件，可以使用file的那种类型
a1.channels.c1.type = memory
#表示用多大的空间
a1.channels.c1.capacity = 1000
#下面表示用事务的空间是多大
a1.channels.c1.transactionCapacity = 100

# 描述和配置source channel sink之间的连接关系,因为source和sink依赖channel来传递数据，所以要分别指定用的是哪个channel。
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

bin/flume-ng agent -c conf -f conf/netcat-logger.conf -n a1 -Dflume.root.logger=INFO,console

测试成功！

三.简单案例

1.采集目录到HDFS

采集需求：服务器的某种特定目录下，会不断产生新的文件，当有新的文件出现，就需要把文件采集到HDFS中

配置文件如下：

##注意：不能往监控目中重复丢同名文件
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /root/logs
a1.sources.r1.fileHeader = true

# Describe the sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.channel = c1
a1.sinks.k1.hdfs.path = /flume/events/%y-%m-%d/%H%M/
a1.sinks.k1.hdfs.filePrefix = events-
#控制文件夹的滚动频率
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute
#控制文件的滚动频率
a1.sinks.k1.hdfs.rollInterval = 3
a1.sinks.k1.hdfs.rollSize = 20
#文件大小维度
a1.sinks.k1.hdfs.rollCount = 5
#event数量维度
a1.sinks.k1.hdfs.batchSize = 1
a1.sinks.k1.hdfs.useLocalTimeStamp = true
#生成的文件类型，默认是Sequencefile，可用DataStream，则为普通文本
a1.sinks.k1.hdfs.fileType = DataStream

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

bin/flume-ng agent -c conf -f conf/spooldir-hdfs.conf -n a1 -Dflume.root.logger=INFO,console

2.采集文件到HDFS

采集需求：比如下午系统使用log4j生成的日志，日志内容不断增加，需要把追加到日志文件中的数据实时采集到hdfs

配置文件如下：

a1.sources = r1
a1.sinks = k1
a1.channels = c1

##注意：不能往监控目中重复丢同名文件
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /root/logs/test.logs
a1.sources.r1.channels = c1

# Describe the sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.channel = c1
a1.sinks.k1.hdfs.path = /flume/events/%y-%m-%d/%H%M/
a1.sinks.k1.hdfs.filePrefix = events-
#控制文件夹的滚动频率
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute
#控制文件的滚动频率
a1.sinks.k1.hdfs.rollInterval = 3
a1.sinks.k1.hdfs.rollSize = 20
#文件大小维度
a1.sinks.k1.hdfs.rollCount = 5
#event数量维度
a1.sinks.k1.hdfs.batchSize = 1
a1.sinks.k1.hdfs.useLocalTimeStamp = true
#生成的文件类型，默认是Sequencefile，可用DataStream，则为普通文本
a1.sinks.k1.hdfs.fileType = DataStream

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

bin/flume-ng agent -c conf -f conf/tail-hdfs.conf -n a1 -Dflume.root.logger=INFO,console

我们写一个脚本，让数据源源不断的写入/root/logs/test.log中

while true ;do echo 'access access ....' >>/root/logs/test.log ; sleep 0.5 ; done