Apache Flume日志收集系统简介

Apache Flume是一个分布式、可靠、可用的系统，用于从大量不同的源有效地收集、聚合、移动大量日志数据进行集中式数据存储。

Flume简介

Flume的核心是Agent，Agent中包含Source、Channel、Sink。Agent是最小的独立运行单位。在Agent中，数据流向为Source->Channel->Sink。

其中，

Source：收集数据，传递给Channel。支持多种收集方式，如RPC、syslog、监控目录。

Channel：数据通道，接收Source的数据并储存，传递给Sink。Channel中的数据在被Sink消费前会一直保存，等Sink成功把数据发送到下一跳Channel或最终目的地后才会删除缓存的数据。

Sink：消费Channel中的数据，传递到下一跳Channel或最终目的地，完成后将数据从Channel中移除。

Flume传输的数据的基本单位是Event，Event同时也是事务操作的基本单位。通常传输的日志内容存储在Event中。Event由可选的header和载有数据的byte array构成。

Flume支持多个Agent相连，形成多级Agent。此时上一级Sink和下一级Source都必须使用Avro协议。

使用多级Flume可以实现日志的聚合，第一层Agent接收日志，第二层Agent统一处理。

Flume支持将流从一个Source扇出到多个Channel。有两种模式的扇出，复制和复用。在复制流程中，事件被发送到所有配置的通道。在复用的情况下，事件仅发送到合格信道的子集。

Avro

Apache Avro是一种数据序列化系统。它是一个基于 RPC 的框架，被 Apache 项目广泛用于数据存储和通信。Avro提供了丰富的数据结构、紧凑快速的二进制数据格式、与动态语言的简单集成。

Avro 依赖于与数据存储在一起的模式。因为没有每个值的开销，实现了轻松而又快速的序列化。当在RPC中使用Avro时，客户端和服务器在连接握手中交换模式。Avro模式是使用JSON定义的，字段在客户端和服务器之间的对应很容易得到解决。

Source

Flume支持多种类型的Source，包括Avro、Thrift、Exec、JMS、Spooling Directory、Taildir、Kafka、NetCat、Sequence Generator、Syslog Sources、HTTP、Stress、Custom、Scribe。

安装后测试时，可以使用NetCat Source监听一个端口，然后Telnet登录该端口输入字符串即可。

程序接入最便捷的方式是让Flume读取现有的日志文件，可以使用如下Source：

Taildir Source：观察指定的文件，并在检测到添加到每个文件的新行后几乎实时地尾随它们。

Spooling Directory Source：监测配置的目录下新增的文件，并将文件中的数据读取出来。需要注意两点：拷贝到 spool 目录下的文件不可以再打开编辑；spool 目录下不可包含相应的子目录。

Exec Source：以运行Linux命令的方式，持续的输出最新的数据，如tail -F文件名指令。

Channel

Flume支持多种类型的Channel，包括Memory、JDBC、Kafka、File、Spillable Memory、Custom、Pseudo Transaction。其中，Memory Channel 可以实现高速的吞吐，但是无法保证数据的完整性；File Channel 是一个持久化的隧道（channel），它持久化所有的事件，并将其存储到磁盘中。

Sink

Flume支持多种类型的Sink，包括HDFS、Hive、Logger、Avro、Thrift、IRC、File Roll、Null、HBase、MorphlineSolr、Elastic Search、Kite Dataset、Kafka、Custom。Sink在设置存储数据时，可以向文件系统、数据库、Hadoop存数据，在日志数据较少时，可以将数据存储在文件系中，并且设定一定的时间间隔保存数据。在日志数据较多时，可以将相应的日志数据存储到Hadoop中，便于日后进行相应的数据分析。

简单使用示例

创建example.conf文件，内容如下

# 配置一个agent，名称为a1，Source、Channel、Sink分别只有1个
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# 配置Source，类型为netcat，监听本机的44444端口
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444
# 配置Sink，类型为logger，输出日志到console
a1.sinks.k1.type = logger
# 配置Channel，类型为memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# 绑定Source、Sink和Channel的对应关系
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

启动Flume agent

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

打开另一个终端，Telnet连接44444端口，并发送数据。

$ telnet localhost 44444
Trying 127.0.0.1...
Connected to localhost.localdomain (127.0.0.1).
Escape character is '^]'.
Hello world! 
OK

12/06/19 15:32:19 INFO source.NetcatSource: Source starting
12/06/19 15:32:19 INFO source.NetcatSource: CreatedserverSocket:sun.nio.ch.ServerSocketChannelImpl[/127.0.0.1:44444]
12/06/19 15:32:34 INFO sink.LoggerSink: Event: { headers:{} body: 48 65 6C6C 6F 20 77 6F 72 6C 64 21 0D Hello world!. }