Flume运行原理

ゝ一世哀愁。 2021-09-12 02:26 432阅读 0赞

1、介绍

flume是由cloudera软件公司产出的可分布式日志收集系统，后与2009年被捐赠了apache软件基金会，为hadoop相关组件之一。尤其近几年随着flume的不断被完善以及升级版本的逐一推出，特别是flume-ng;同时flume内部的各种组件不断丰富，用户在开发的过程中使用的便利性得到很大的改善，现已成为apache top项目之一.

2. 架构  
Flume的架构主要有一下几个核心概念：  
  
Event：一个数据单元，带有一个可选的消息头  
Flow：Event从源点到达目的点的迁移的抽象  
Client：操作位于源点处的Event，将其发送到Flume Agent  
Agent：一个独立的Flume进程，包含组件Source、Channel、Sink  
Source：用来消费传递到该组件的Event  
Channel：中转Event的一个临时存储，保存有Source组件传递过来的Event  
Sink：从Channel中读取并移除Event，将Event传递到Flow Pipeline中的下一个Agent（如果有的话）

3. Flume的结构  
(1). flume的外部结构

Flume 的核心是把数据从数据源收集过来，再送到目的地。为了保证输送一定成功，在送到目的地之前，会先缓存数据，待数据真正到达目的地后，删除自己缓存的数据。  
  
Flume 传输的数据的基本单位是 Event，如果是文本文件，通常是一行记录，这也是事务的基本单位。Event 从 Source，流向 Channel，再到 Sink，本身为一个 byte 数组，并可携带 headers 信息。Event 代表着一个数据流的最小完整单元，从外部数据源来，向外部的目的地去。  
  
Flume 运行的核心是 Agent。它是一个完整的数据收集工具，含有三个核心组件，分别是 source、channel、sink。通过这些组件，Event 可以从一个地方流向另一个地方，如下图所示。

![20170319162240694][]

source 可以接收外部源发送过来的数据。不同的 source，可以接受不同的数据格式。比如有目录池(spooling directory)数据源，可以监控指定文件夹中的新文件变化，如果目录中有文件产生，就会立刻读取其内容。  
channel 是一个存储地，接收 source 的输出，直到有 sink 消费掉 channel 中的数据。channel 中的数据直到进入到下一个channel中或者进入终端才会被删除。当 sink 写入失败后，可以自动重启，不会造成数据丢失，因此很可靠。  
sink 会消费 channel 中的数据，然后送给外部源或者其他 source。如数据可以写入到 HDFS 或者 HBase 中。  
flume 允许多个 agent 连在一起，形成前后相连的多级跳。

![20170319162426401][]

(2). Flume 事件  
  
事件作为Flume内部数据传输的最基本单元.它是由一个转载数据的字节数组(该数据组是从数据源接入点传入，并传输给传输器，也就是HDFS/HBase)和一个可选头部构成.  
  
典型的Flume 事件如下面结构所示：

![20170319162530430][]

4、核心组件

（1）soure

Client端操作消费数据的来源，Flume 支持 Avro，log4j，syslog 和 http post(body为json格式)。可以让应用程序同已有的Source直接打交道，如AvroSource，SyslogTcpSource。也可以 写一个 Source，以 IPC 或 RPC 的方式接入自己的应用，Avro和 Thrift 都可以(分别有 NettyAvroRpcClient 和 ThriftRpcClient 实现了 RpcClient接口)，其中 Avro 是默认的 RPC 协议。具体代码级别的 Client 端数据接入，可以参考官方手册。

Flume Source 支持的类型：

<table style="box-sizing: border-box; border-collapse: collapse; border-spacing: 0px; margin: 1.5em 0px; width: 825px;"> 
 <thead style="box-sizing: border-box;"> 
  <tr style="box-sizing: border-box;"> 
   <th style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal; background-image: initial; background-attachment: initial; background-size: initial; background-origin: initial; background-clip: initial; background-position: initial; background-repeat: initial;">Source类型</th> 
   <th style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal; background-image: initial; background-attachment: initial; background-size: initial; background-origin: initial; background-clip: initial; background-position: initial; background-repeat: initial;">说明</th> 
  </tr> 
 </thead> 
 <tbody style="box-sizing: border-box;"> 
  <tr style="box-sizing: border-box;"> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">Avro Source</td> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">支持Avro协议（实际上是Avro RPC），内置支持</td> 
  </tr> 
  <tr style="box-sizing: border-box;"> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">Thrift Source</td> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">支持Thrift协议，内置支持</td> 
  </tr> 
  <tr style="box-sizing: border-box;"> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">&nbsp;</td> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">Exec Source | 基于Unix的command在标准输出上生产数据</td> 
  </tr> 
  <tr style="box-sizing: border-box;"> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">JMS Source</td> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">从JMS系统（消息、主题）中读取数据，ActiveMQ已经测试过</td> 
  </tr> 
  <tr style="box-sizing: border-box;"> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">Spooling Directory Source</td> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">监控指定目录内数据变更</td> 
  </tr> 
  <tr style="box-sizing: border-box;"> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">Twitter 1% firehose Source</td> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">通过API持续下载Twitter数据，试验性质</td> 
  </tr> 
  <tr style="box-sizing: border-box;"> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">Netcat Source</td> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">监控某个端口，将流经端口的每一个文本行数据作为Event输入</td> 
  </tr> 
  <tr style="box-sizing: border-box;"> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">Sequence Generator Source</td> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">序列生成器数据源，生产序列数据</td> 
  </tr> 
  <tr style="box-sizing: border-box;"> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">Syslog Sources</td> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">读取syslog数据，产生Event，支持UDP和TCP两种协议</td> 
  </tr> 
  <tr style="box-sizing: border-box;"> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">HTTP Source</td> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">基于HTTP POST或GET方式的数据源，支持JSON、BLOB表示形式</td> 
  </tr> 
  <tr style="box-sizing: border-box;"> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">Legacy Sources</td> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">兼容老的Flume OG中Source（0.9.x版本）</td> 
  </tr> 
 </tbody> 
</table>

如果上面都不符合你的要求，可以自己写一上source

![20170319163030566][]

然后指明使用自己的source:

![20170319163136667][]

(2)channel

当前有几个 channel 可供选择，分别是 Memory Channel, JDBC Channel , File Channel，Psuedo Transaction Channel。比较常见的是前三种 channel。  
  
MemoryChannel 可以实现高速的吞吐，但是无法保证数据的完整性。  
MemoryRecoverChannel 在官方文档的建议上已经建义使用FileChannel来替换。  
FileChannel保证数据的完整性与一致性。在具体配置FileChannel时，建议FileChannel设置的目录和程序日志文件保存的目录设成不同的磁盘，以便提高效率。  
File Channel 是一个持久化的隧道（channel），它持久化所有的事件，并将其存储到磁盘中。因此，即使 Java 虚拟机当掉，或者操作系统崩溃或重启，再或者事件没有在管道中成功地传递到下一个代理（agent），这一切都不会造成数据丢失。Memory Channel 是一个不稳定的隧道，其原因是由于它在内存中存储所有事件。如果 java 进程死掉，任何存储在内存的事件将会丢失。另外，内存的空间收到 RAM大小的限制,而 File Channel 这方面是它的优势，只要磁盘空间足够，它就可以将所有事件数据存储到磁盘上。  
  
Flume Channel 支持的类型：

如果处理数据很大的话，Spillable Memory Channel建议使用这个，它将内存和磁盘结合起来了。数据优先存储到内存，内存不够再使用磁盘。

数据量小的话，就用MemoryChannel

（3）sinkSink

在设置存储数据时，可以向文件系统、数据库、hadoop存数据，在日志数据较少时，可以将数据存储在文件系中，并且设定一定的时间间隔保存数据。在日志数据较多时，可以将相应的日志数据存储到Hadoop中，便于日后进行相应的数据分析。

Flume Sink支持的类型

<table style="box-sizing: border-box; border-collapse: collapse; border-spacing: 0px; margin: 1.5em 0px; width: 825px;"> 
 <thead style="box-sizing: border-box;"> 
  <tr style="box-sizing: border-box;"> 
   <th style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal; background: rgb(243, 243, 243);">Sink类型</th> 
   <th style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal; background: rgb(243, 243, 243);">说明</th> 
  </tr> 
 </thead> 
 <tbody style="box-sizing: border-box;"> 
  <tr style="box-sizing: border-box;"> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">HDFS Sink</td> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">数据写入HDFS</td> 
  </tr> 
  <tr style="box-sizing: border-box;"> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">Logger Sink</td> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">数据写入日志文件</td> 
  </tr> 
  <tr style="box-sizing: border-box;"> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">Avro Sink</td> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">数据被转换成Avro Event，然后发送到配置的RPC端口上</td> 
  </tr> 
  <tr style="box-sizing: border-box;"> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">Thrift Sink</td> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">数据被转换成Thrift Event，然后发送到配置的RPC端口上</td> 
  </tr> 
  <tr style="box-sizing: border-box;"> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">IRC Sink</td> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">数据在IRC上进行回放</td> 
  </tr> 
  <tr style="box-sizing: border-box;"> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">File Roll Sink</td> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">存储数据到本地文件系统</td> 
  </tr> 
  <tr style="box-sizing: border-box;"> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">Null Sink</td> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">丢弃到所有数据</td> 
  </tr> 
  <tr style="box-sizing: border-box;"> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">HBase Sink</td> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">数据写入HBase数据库</td> 
  </tr> 
  <tr style="box-sizing: border-box;"> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">Morphline Solr Sink</td> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">数据发送到Solr搜索服务器（集群）</td> 
  </tr> 
  <tr style="box-sizing: border-box;"> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">ElasticSearch Sink</td> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">数据发送到Elastic Search搜索服务器（集群）</td> 
  </tr> 
  <tr style="box-sizing: border-box;"> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">Kite Dataset Sink</td> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">写数据到Kite Dataset，试验性质的</td> 
  </tr> 
  <tr style="box-sizing: border-box;"> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">Custom Sink</td> 
   <td style="box-sizing: border-box; padding: 5px 8px; border: 1px solid rgb(230, 230, 230); word-break: normal;">自定义Sink实现</td> 
  </tr> 
 </tbody> 
</table>

也可以自定义sink:

![20170319163733125][]

然后是使用:

![20170319163753453][]

4、安装与使用

（1）安装

安装很简单，下个tar包。找个机器。将文件解压即可。

（2）使用

一个典型的flume启动时加载的properties文件，在文件中指明sources/channels/sinks的关系

![20170319164613821][]

![20170319164621634][]

启动脚本：

![20170319164127095][]

\--conf指明配置文件的目录，这里一般还会放一些hive-site.xml/hbase-site.xml等和hadoop环境相关的xml文件

\-f指明使用的sink/source/channel的文件

\-C libs 指明依赖的jar包

\-n a1 指明agent的名字

![20170319164833223][]

参考文章：

https://segmentfault.com/a/1190000002532284

http://www.cnblogs.com/gongxijun/p/5656778.html

https://segmentfault.com/a/1190000002532284

http://www.cnblogs.com/gongxijun/p/5656778.html

[20170319162240694]: /images/20210811/64ae6e8a53da428daa1792a848bb56fd.png
[20170319162426401]: /images/20210811/a9b768a213c2410ea33ccaaaffdd9c7c.png
[20170319162530430]: /images/20210811/ccf9e87a184f4c598b1e69061283ea0c.png
[20170319163030566]: /images/20210811/a3e7020cb2844f80840e646990828cfe.png
[20170319163136667]: /images/20210811/fb4f24a67f5d46a3b347c2b3e8e3145d.png
[20170319163733125]: /images/20210811/f2013f34efd54efd8439e976bf6c6455.png
[20170319163753453]: /images/20210811/94876927df8b4fa9b3119e25629b39d9.png
[20170319164613821]: /images/20210811/e1171eb6731340dbac82c1c9cea768ea.png
[20170319164621634]: /images/20210811/fb94144888684217967bc6e44a87fe08.png
[20170319164127095]: /images/20210811/6b94b9fe5e7249fc8a1848df8aa7ada9.png
[20170319164833223]: /images/20210811/8272954ca41e4bd898089c644ebfb782.png