大数据（十二）Flume

2020-03-15 阅读量

× 文章目录

1. Quick Guide

Flume是一个分布式、可靠、高可用的海量日志聚合系统，支持在系统中定制各类数据发送方，用于收集数据；同时Flume提供对数据的简单处理，并写到各种数据接收方的能力。

本文章主要介绍 Flume 和在centos环境下如何安装。

Quick Guide

一、背景

Hadoop业务的整体开发流程：

　　从Hadoop的业务开发流程图中可以看出，在大数据的业务处理过程中，对于数据的采集是十分重要的一步，也是不可避免的一步.
　　许多公司的平台每天会产生大量的日志（一般为流式数据，如搜索引擎的pv和查询等），处理这些日志需要特定的日志系统，一般而言，这些系统需要具有以下特征：

构建应用系统和分析系统的桥梁，并将它们之间的关联解耦；
支持近实时的在线分析系统和类似于Hadoop之类的离线分析系统；
具有高可扩展性。即：当数据量增加时，可以通过增加节点进行水平扩展。

开源的日志系统，包括facebook的scribe，apache的chukwa，linkedin的kafka和cloudera的flume等。

二、Flume的简介

　　flume 作为 cloudera 开发的实时日志收集系统，受到了业界的认可与广泛应用。Flume 初始的发行版本目前被统称为 Flume OG（original generation），属于 cloudera。
　　但随着 FLume 功能的扩展，Flume OG 代码工程臃肿、核心组件设计不合理、核心配置不标准等缺点暴露出来，尤其是在 Flume OG 的最后一个发行版本 0.9.4. 中，日志传输不稳定的现象尤为严重，为了解决这些问题，2011 年 10 月 22 号，cloudera 完成了 Flume-728，对 Flume 进行了里程碑式的改动：重构核心组件、核心配置以及代码架构，重构后的版本统称为 Flume NG（next generation）；改动的另一原因是将 Flume 纳入 apache 旗下，cloudera Flume 改名为 Apache Flume。
　　Flume 在0.9.x and 1.x之间有较大的架构调整，1.x版本之后的改称Flume NG，0.9.x的称为Flume OG。
　　Flume目前只有Linux系统的启动脚本，没有Windows环境的启动脚本。　　

三、Flume NG的介绍

3.1 flume特点

　　flume的数据流由事件(Event)贯穿始终。事件是Flume的基本数据单位，它携带日志数据(字节数组形式)并且携带有头信息，这些Event由Agent外部的Source生成，当Source捕获事件后会进行特定的格式化，然后Source会把事件推入(单个或多个)Channel中。你可以把Channel看作是一个缓冲区，它将保存事件直到Sink处理完该事件。Sink负责持久化日志或者把事件推向另一个Source。

　（1）flume的可靠性
　　当节点出现故障时，日志能够被传送到其他节点上而不会丢失。Flume提供了三种级别的可靠性保障，从强到弱依次分别为：end-to-end（收到数据agent首先将event写到磁盘上，当数据传送成功后，再删除；如果数据发送失败，可以重新发送。），Store on failure（这也是scribe采用的策略，当数据接收方crash时，将数据写到本地，待恢复后，继续发送），Besteffort（数据发送到接收方后，不会进行确认）。

　（2）flume的可恢复性
　　依靠Channel，推荐使用FileChannel，事件持久化在本地文件系统里(性能较差)。

3.2 Flume的核心概念

Client：Client生产数据，运行在一个独立的线程。
Event：一个数据单元，消息头和消息体组成。（Events可以是日志记录、 avro 对象等。）
Flow： Event从源点到达目的点的迁移的抽象。
Agent：一个独立的Flume进程，包含组件Source、 Channel、 Sink。（Agent使用JVM 运行Flume。每台机器运行一个agent，但是可以在一个agent中包含多个sources和sinks。）
Source(源端数据采集)：数据收集组件。（source从Client收集数据，传递给Channel）
Channel(临时存储聚合数据)：中转Event的一个临时存储，保存由Source组件传递过来的Event。（Channel连接 sources 和 sinks ，这个有点像一个队列。）
Sink（移动数据到目标端）：从Channel中读取并移除Event，将Event传递到FlowPipeline中的下一个Agent（如果有的话）（Sink从Channel收集数据，运行在一个独立线程。）

3.3 Flume NG的体系结构

Flume 运行的核心是 Agent。Flume以agent为最小的独立运行单位。一个agent就是一个JVM。它是一个完整的数据收集工具，含有三个核心组件，分别是source、 channel、 sink。通过这些组件， Event 可以从一个地方流向另一个地方，如下图所示。
　

3.4 Source

　　Source是数据的收集端，负责将数据捕获后进行特殊的格式化，将数据封装到事件（event）里，然后将事件推入Channel中。
　　Flume提供了各种source的实现，包括Avro Source、Exce Source、Spooling Directory Source、NetCat Source、Syslog Source、Syslog TCP Source、Syslog UDP Source、HTTP Source、HDFS Source，etc。如果内置的Source无法满足需要， Flume还支持自定义Source。
　　

3.5 Channel

　　Channel是连接Source和Sink的组件，大家可以将它看做一个数据的缓冲区（数据队列），它可以将事件暂存到内存中也可以持久化到本地磁盘上，直到Sink处理完该事件。
　　Flume对于Channel，则提供了Memory Channel、JDBC Chanel、File Channel，etc。
　　MemoryChannel可以实现高速的吞吐，但是无法保证数据的完整性。
　　MemoryRecoverChannel在官方文档的建议上已经建义使用FileChannel来替换。
　　FileChannel保证数据的完整性与一致性。在具体配置不现的FileChannel时，建议FileChannel设置的目录和程序日志文件保存的目录设成不同的磁盘，以便提高效率。　　

3.6 Sink

　　Flume Sink取出Channel中的数据，进行相应的存储文件系统，数据库，或者提交到远程服务器。
　　Flume也提供了各种sink的实现，包括HDFS sink、Logger sink、Avro sink、File Roll sink、Null sink、HBase sink，etc。
　　Flume Sink在设置存储数据时，可以向文件系统中，数据库中，hadoop中储数据，在日志数据较少时，可以将数据存储在文件系中，并且设定一定的时间间隔保存数据。在日志数据较多时，可以将相应的日志数据存储到Hadoop中，便于日后进行相应的数据分析。
　　

四、Flume的部署类型

单一流程：

多代理流程（多个agent顺序连接）：

流的合并（多个Agent的数据汇聚到同一个Agent ）：

多路复用流（多级流）：

load balance功能:

五、Flume的安装

0 .从官网下载或者使用下面的命令下载

1 2	cd /hadoop/software wget http://mirrors.hust.edu.cn/apache/flume/1.8.0/apache-flume-1.8.0-bin.tar.gz

1.解压并重命名

1
2
3

tar -zxvf apache-flume-1.8.0-bin.tar.gz -C /hadoop/install/
cd /hadoop/install
mv apache-flume-1.8.0-bin apache-flume-1.8.0

2.配置环境变量

1	vi /etc/profile

添加内容:

1
2
3

export FLUME_HOME=/hadoop/install/apache-flume-1.8.0
export FLUME_CONF_DIR=$FLUME_HOME/conf
export PATH=$PATH:$JAVA_HOME/bin:$HADOOP_HOME/bin:$HIVE_HOME/bin:$ZK_HOME/bin:$HBASE_HOME/bin:$FLUME_HOME/bin

3.修改配置文件

3.1.修改flume-env.sh

cd /hadoop/install/apache-flume-1.8.0/conf
# 拷贝出模板
cp flume-env.sh.template flume-env.sh
# 修改配置文件
vi flume-env.sh

3.2.修改flume-conf

1 2	cp flume-conf.properties.template flume-conf.properties vi flume-conf.properties

添加agent、source、channel和sink的相关信息

# The configuration file needs to define the sources, the channels and the sinks.
# Sources, channels and sinks are defined per agent, in this case called 'a1'
a1.sources=r1
a1.channels=c1
a1.sinks=k1
 
#Describe/configure the source
a1.sources.r1.type=netcat
a1.sources.r1.bind=192.168.52.110
a1.sources.r1.port=44444
a1.sources.r1.channels=c1

#Describe the sink
a1.sinks.k1.type=logger
a1.sinks.k1.channel=c1
 
#Describe the channel
a1.channels.c1.type=memory
a1.channels.c1.capacity=1000
a1.channels.c1.transcationCapacity=100

配置文件说明：

| 配置项名称                 | 作用                                                         | 举例                                                      |
| ---------------- | ------------------------ | ------------------ |
| agent1                     | flume节点名称，启动时通过参数命令-name指定                   |                                                           |
| agent1.sources             | 监听的源,可以有多个，空格隔开即可，用于收集数据并发送到channel | agent1.sources=s1 s2                                      |
| agent1.channels            | 临时通道，可以有多个，空格隔开，存放source收集的数据，sink从这里读取数据 | agent1.channels=c1                                        |
| agent1.sinks               | 接收器，可以有多个，空格隔开，从channel读取数据，并发送给目标 （比如kafka 或者hdfs或者另一个flume） | agent1.sinks=k1                                           |
| agent.sources.s1.type      | 源的类型，s1是源名称，可以是目录、日志文件、或者监听端口等。 常用的source的类型包括avro、exec、netcat、spooldir和syslog等. 具体请参考官网[http://flume.apache.org/Flume...](http://flume.apache.org/FlumeUserGuide.html#flume-sources) | agent.sources.s1.type=spooldir agent.sources.s2.type=avro |
| agent1.sources.s1.channels | 监听数据要写入的channel名称                                  | agent1.sources.s1.channels=c1                             |
| agent1.channels.c1.type    | 通道的类型，c1为通道名称，此处memory表示为常用的memory-channel， 同时也有其他类型的channel，如JDBC、file-channel、custom-channel等， 具体请参考官网[http://flume.apache.org/Flume...](http://flume.apache.org/FlumeUserGuide.html#flume-channels) | agent1.channels.c1.type=memory                            |
| agent1.sinks.k1.type       | 接收器类型，k1为接收器名称，此处logger表示直接写入日志， 常用的包括avro、logger、HDFS、Hbase以及file-roll等， 具体参考官网[http://flume.apache.org/Flume...](http://flume.apache.org/FlumeUserGuide.html#flume-sinks) | agent1.sinks.k1.type=logger                               |

4.启动

1	./bin/flume-ng agent --conf <conf路径> --conf-file <flume-conf.properties路径> --name <agent名称> -Dflume.root.logger=INFO,console

启动命令参数说明：

参数	作用	举例
–conf 或 -c	指定配置文件夹，包含flume-env.sh和log4j的配置文件	–conf conf
–conf-file 或 -f	配置文件地址	–conf-file conf/flume.conf
–name 或 -n	agent(flume节点)名称	–name agent1

1
2
3

cd /hadoop/install/apache-flume-1.8.0
# 启动命令:
./bin/flume-ng agent --conf ./conf --conf-file ./conf/flume-conf.properties --name a1 -Dflume.root.logger=INFO,console

5.测试是否运行成功：

1	telnet 192.168.52.110 44444

六、测试收集日志到HDFS

1.新建配置2

1
2
3

cd /hadoop/install/apache-flume-1.8.0/conf
cp flume-conf.properties.template flume-conf2.properties
vi flume-conf2.properties

2.配置2内容

a1.sources = r1
a1.sinks = k1
a1.channels = c1
a1.sources.r1.type = exec
a1.sources.r1.channels = c1
a1.sources.r1.command = tail -F /hadoop/install/hadoop/logs/hadoop-namenode-b393a04554e1.log
a1.sinks.k1.type = hdfs
a1.sinks.k1.channel = c1
a1.sinks.k1.hdfs.path = hdfs://master:54310/class12/out_flume
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.rollSize = 4000000
a1.sinks.k1.hdfs.rollCount = 0
a1.sinks.k1.hdfs.writeFormat = Text
a1.sinks.k1.hdfs.fileType = DataStream
a1.sinks.k1.hdfs.batchSize = 10
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

3.使用配置2启动

1
2
3

cd /hadoop/install/apache-flume-1.8.0

./bin/flume-ng agent --conf ./conf/ --conf-file ./conf/flume-conf2.properties --name a1 -Dflume.root.logger=INFO,console