Apache Flume日志收集系统简介

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

Flume简介

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

Apache Flume日志收集系统简介

其中,

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

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

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

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

Apache Flume日志收集系统简介

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

Apache Flume日志收集系统简介

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

Apache Flume日志收集系统简介

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

Apache Flume日志收集系统简介

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文件,内容如下

  1. # 配置一个agent,名称为a1,Source、Channel、Sink分别只有1个
  2. a1.sources = r1
  3. a1.sinks = k1
  4. a1.channels = c1
  5. # 配置Source,类型为netcat,监听本机的44444端口
  6. a1.sources.r1.type = netcat
  7. a1.sources.r1.bind = localhost
  8. a1.sources.r1.port = 44444
  9. # 配置Sink,类型为logger,输出日志到console
  10. a1.sinks.k1.type = logger
  11. # 配置Channel,类型为memory
  12. a1.channels.c1.type = memory
  13. a1.channels.c1.capacity = 1000
  14. a1.channels.c1.transactionCapacity = 100
  15. # 绑定Source、Sink和Channel的对应关系
  16. a1.sources.r1.channels = c1
  17. a1.sinks.k1.channel = c1

启动Flume agent

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

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

  1. $ telnet localhost 44444
  2. Trying 127.0.0.1...
  3. Connected to localhost.localdomain (127.0.0.1).
  4. Escape character is '^]'.
  5. Hello world! <ENTER>
  6. OK

可看到Flume在console输出如下内容

  1. 12/06/19 15:32:19 INFO source.NetcatSource: Source starting
  2. 12/06/19 15:32:19 INFO source.NetcatSource: Created serverSocket:sun.nio.ch.ServerSocketChannelImpl[/127.0.0.1:44444]
  3. 12/06/19 15:32:34 INFO sink.LoggerSink: Event: { headers:{} body: 48 65 6C 6C 6F 20 77 6F 72 6C 64 21 0D          Hello world!. }
上一篇:Flume 和 kafka的区别和对比


下一篇:Notes for Studying Django