Flume原理与应用

妖狐艹你老母 2022-05-25 03:35 244阅读 0赞

# 1、简介 #

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

Flume的使用不仅限于日志数据聚合。由于数据源是可定制的，所以可以使用Flume来传输大量的事件数据，包括但不限于网络流量数据、社交媒体生成的数据、电子邮件消息和几乎所有可能的数据源。

# 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；

## 2.1 数据流 ##

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

Flume 传输的数据的基本单位是 Event，如果是文本文件，通常是一行记录，这也是事务的基本单位。Event 从 Source，流向 Channel，再到 Sink，本身为一个 byte 数组，并可携带 headers 信息。Event 代表着一个数据流的最小完整单元，从外部数据源来，向外部的目的地去。

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

![20180507230026472][]

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

### 2.2 核心组件 ###

#### 2.2.1 source ####

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

对现有程序改动最小的使用方式是使用是直接读取程序原来记录的日志文件，基本可以实现无缝接入，不需要对现有程序进行任何改动。 对于直接读取文件 Source,有两种方式：

*  ExecSource: 以运行 Linux 命令的方式，持续的输出最新的数据，如 `tail -F 文件名` 指令，在这种方式下，取的文件名必须是指定的。 ExecSource 可以实现对日志的实时收集，但是存在Flume不运行或者指令执行出错时，将无法收集到日志数据，无法保证日志数据的完整性。
 *  SpoolSource: 监测配置的目录下新增的文件，并将文件中的数据读取出来。需要注意两点：拷贝到 spool 目录下的文件不可以再打开编辑；spool 目录下不可包含相应的子目录。

SpoolSource 虽然无法实现实时的收集数据，但是可以使用以分钟的方式分割文件，趋近于实时。

如果应用无法实现以分钟切割日志文件的话， 可以两种收集方式结合使用。 在实际使用的过程中，可以结合 log4j 使用，使用 log4j的时候，将 log4j 的文件分割机制设为1分钟一次，将文件拷贝到spool的监控目录。

log4j 有一个 TimeRolling 的插件，可以把 log4j 分割文件到 spool 目录。基本实现了实时的监控。Flume 在传完文件之后，将会修改文件的后缀，变为 .COMPLETED（后缀也可以在配置文件中灵活指定）。

Flume Source 支持的类型：

<table style="border-spacing:0px;width:852px;text-align:center;color:rgb(51,51,51);font-family:'-apple-system', 'SF UI Text', Arial, 'PingFang SC', 'Hiragino Sans GB', 'Microsoft YaHei', 'WenQuanYi Micro Hei', sans-serif, SimHei, SimSun;font-size:14px;background-color:rgb(255,255,255);"> 
 <thead> 
  <tr> 
   <th align="left" style="margin:0px;vertical-align:middle;">Source类型</th> 
   <th align="left" style="margin:0px;vertical-align:middle;">说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left" style="margin:0px;vertical-align:middle;">Avro Source</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">支持Avro协议（实际上是Avro RPC），内置支持</td> 
  </tr> 
  <tr> 
   <td align="left" style="margin:0px;vertical-align:middle;">Thrift Source</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">支持Thrift协议，内置支持</td> 
  </tr> 
  <tr> 
   <td align="left" style="margin:0px;vertical-align:middle;">Exec Source</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">基于Unix的command在标准输出上生产数据</td> 
  </tr> 
  <tr> 
   <td align="left" style="margin:0px;vertical-align:middle;">JMS Source</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">从JMS系统（消息、主题）中读取数据，ActiveMQ已经测试过</td> 
  </tr> 
  <tr> 
   <td align="left" style="margin:0px;vertical-align:middle;">Spooling Directory Source</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">监控指定目录内数据变更</td> 
  </tr> 
  <tr> 
   <td align="left" style="margin:0px;vertical-align:middle;">Twitter 1% firehose Source</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">通过API持续下载Twitter数据，试验性质</td> 
  </tr> 
  <tr> 
   <td align="left" style="margin:0px;vertical-align:middle;">Netcat Source</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">监控某个端口，将流经端口的每一个文本行数据作为Event输入</td> 
  </tr> 
  <tr> 
   <td align="left" style="margin:0px;vertical-align:middle;">Sequence Generator Source</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">序列生成器数据源，生产序列数据</td> 
  </tr> 
  <tr> 
   <td align="left" style="margin:0px;vertical-align:middle;">Syslog Sources</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">读取syslog数据，产生Event，支持UDP和TCP两种协议</td> 
  </tr> 
  <tr> 
   <td align="left" style="margin:0px;vertical-align:middle;">HTTP Source</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">基于HTTP POST或GET方式的数据源，支持JSON、BLOB表示形式</td> 
  </tr> 
  <tr> 
   <td align="left" style="margin:0px;vertical-align:middle;">Legacy Sources</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">兼容老的Flume OG中Source（0.9.x版本）</td> 
  </tr> 
 </tbody> 
</table>

#### 2.2.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 支持的类型：

#### 2.2.3 sink ####

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

Flume Sink支持的类型

<table style="border-spacing:0px;width:852px;text-align:center;color:rgb(51,51,51);font-family:'-apple-system', 'SF UI Text', Arial, 'PingFang SC', 'Hiragino Sans GB', 'Microsoft YaHei', 'WenQuanYi Micro Hei', sans-serif, SimHei, SimSun;font-size:14px;background-color:rgb(255,255,255);"> 
 <thead> 
  <tr> 
   <th align="left" style="margin:0px;vertical-align:middle;">Sink类型</th> 
   <th align="left" style="margin:0px;vertical-align:middle;">说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left" style="margin:0px;vertical-align:middle;">HDFS Sink</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">数据写入HDFS</td> 
  </tr> 
  <tr> 
   <td align="left" style="margin:0px;vertical-align:middle;">Logger Sink</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">数据写入日志文件</td> 
  </tr> 
  <tr> 
   <td align="left" style="margin:0px;vertical-align:middle;">Avro Sink</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">数据被转换成Avro Event，然后发送到配置的RPC端口上</td> 
  </tr> 
  <tr> 
   <td align="left" style="margin:0px;vertical-align:middle;">Thrift Sink</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">数据被转换成Thrift Event，然后发送到配置的RPC端口上</td> 
  </tr> 
  <tr> 
   <td align="left" style="margin:0px;vertical-align:middle;">IRC Sink</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">数据在IRC上进行回放</td> 
  </tr> 
  <tr> 
   <td align="left" style="margin:0px;vertical-align:middle;">File Roll Sink</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">存储数据到本地文件系统</td> 
  </tr> 
  <tr> 
   <td align="left" style="margin:0px;vertical-align:middle;">Null Sink</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">丢弃到所有数据</td> 
  </tr> 
  <tr> 
   <td align="left" style="margin:0px;vertical-align:middle;">HBase Sink</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">数据写入HBase数据库</td> 
  </tr> 
  <tr> 
   <td align="left" style="margin:0px;vertical-align:middle;">Morphline Solr Sink</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">数据发送到Solr搜索服务器（集群）</td> 
  </tr> 
  <tr> 
   <td align="left" style="margin:0px;vertical-align:middle;">ElasticSearch Sink</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">数据发送到Elastic Search搜索服务器（集群）</td> 
  </tr> 
  <tr> 
   <td align="left" style="margin:0px;vertical-align:middle;">Kite Dataset Sink</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">写数据到Kite Dataset，试验性质的</td> 
  </tr> 
  <tr> 
   <td align="left" style="margin:0px;vertical-align:middle;">Custom Sink</td> 
   <td align="left" style="margin:0px;vertical-align:middle;">自定义Sink实现</td> 
  </tr> 
 </tbody> 
</table>

更多sink的内容可以参考[官方手册][Link 1]。

### 2.3 可靠性 ###

Flume 使用事务性的方式保证传送Event整个过程的可靠性。Sink 必须在 Event 被存入 Channel 后，或者，已经被传达到下一站agent里，又或者，已经被存入外部数据目的地之后，才能把 Event 从 Channel 中 remove 掉。这样数据流里的 event 无论是在一个 agent 里还是多个 agent 之间流转，都能保证可靠，因为以上的事务保证了 event 会被成功存储起来。而 Channel 的多种实现在可恢复性上有不同的保证。也保证了 event 不同程度的可靠性。比如 Flume 支持在本地保存一份文件 channel 作为备份，而memory channel 将 event 存在内存 queue 里，速度快，但丢失的话无法恢复。

## 3. 使用场景 ##

下面，根据官网文档，我们展示几种Flow Pipeline，各自适应于什么样的应用场景：

*  多个 agent 顺序连接：

![20180507230418199][]

可以将多个Agent顺序连接起来，将最初的数据源经过收集，存储到最终的存储系统中。这是最简单的情况，一般情况下，应该控制这种顺序连接的Agent的数量，因为数据流经的路径变长了，如果不考虑failover的话，出现故障将影响整个Flow上的Agent收集服务。

*  多个Agent的数据汇聚到同一个Agent:

![20180507230458229][]

这种情况应用的场景比较多，比如要收集Web网站的用户行为日志，Web网站为了可用性使用的负载均衡的集群模式，每个节点都产生用户行为日志，可以为每个节点都配置一个Agent来单独收集日志数据，然后多个Agent将数据最终汇聚到一个用来存储数据存储系统，如HDFS上。

*  多路（Multiplexing）Agent

![20180507230509281][]

这种模式，有两种方式，一种是用来复制（Replication），另一种是用来分流（Multiplexing）。Replication方式，可以将最前端的数据源复制多份，分别传递到多个channel中，每个channel接收到的数据都是相同的。

配置格式示例如下：

# List the sources, sinks and channels for the agent
    <Agent>.sources = <Source1>
    <Agent>.sinks = <Sink1> <Sink2>
    <Agent>.channels = <Channel1> <Channel2>
    
    # set list of channels for source (separated by space)
    <Agent>.sources.<Source1>.channels = <Channel1> <Channel2>
    
    # set channel for sinks
    <Agent>.sinks.<Sink1>.channel = <Channel1>
    <Agent>.sinks.<Sink2>.channel = <Channel2>
    
    <Agent>.sources.<Source1>.selector.type = replicating

上面指定了selector的type的值为replication，其他的配置没有指定，使用的Replication方式，Source1会将数据分别存储到Channel1和Channel2，这两个channel里面存储的数据是相同的，然后数据被传递到Sink1和Sink2。

Multiplexing方式，selector可以根据header的值来确定数据传递到哪一个channel，配置格式，如下所示：

# Mapping for multiplexing selector
    <Agent>.sources.<Source1>.selector.type = multiplexing
    <Agent>.sources.<Source1>.selector.header = <someHeader>
    <Agent>.sources.<Source1>.selector.mapping.<Value1> = <Channel1>
    <Agent>.sources.<Source1>.selector.mapping.<Value2> = <Channel1> <Channel2>
    <Agent>.sources.<Source1>.selector.mapping.<Value3> = <Channel2>
    #...
    
    <Agent>.sources.<Source1>.selector.default = <Channel2>

上面selector的type的值为multiplexing，同时配置selector的header信息，还配置了多个selector的mapping的值，即header的值：如果header的值为Value1、Value2，数据从Source1路由到Channel1；如果header的值为Value2、Value3，数据从Source1路由到Channel2。

*  实现load balance功能

![这里写图片描述][SouthEast]

Load balancing Sink Processor能够实现load balance功能，上图Agent1是一个路由节点，负责将Channel暂存的Event均衡到对应的多个Sink组件上，而每个Sink组件分别连接到一个独立的Agent上，示例配置，如下所示：

a1.sinkgroups = g1
    a1.sinkgroups.g1.sinks = k1 k2 k3
    a1.sinkgroups.g1.processor.type = load_balance
    a1.sinkgroups.g1.processor.backoff = true
    a1.sinkgroups.g1.processor.selector = round_robin
    a1.sinkgroups.g1.processor.selector.maxTimeOut=10000

*  实现failover能

Failover Sink Processor能够实现failover功能，具体流程类似load balance，但是内部处理机制与load balance完全不同：Failover Sink Processor维护一个优先级Sink组件列表，只要有一个Sink组件可用，Event就被传递到下一个组件。如果一个Sink能够成功处理Event，则会加入到一个Pool中，否则会被移出Pool并计算失败次数，设置一个惩罚因子，示例配置如下所示：

a1.sinkgroups = g1
    a1.sinkgroups.g1.sinks = k1 k2 k3
    a1.sinkgroups.g1.processor.type = failover
    a1.sinkgroups.g1.processor.priority.k1 = 5
    a1.sinkgroups.g1.processor.priority.k2 = 7
    a1.sinkgroups.g1.processor.priority.k3 = 6
    a1.sinkgroups.g1.processor.maxpenalty = 20000

## 4. 安装 ##

（1）在数据源所在的节点上进行解压安装

（2）配置环境变量

export FLUME\_HOME=/usr/local/src/flume-1.8.0

export PATH=$PATH:$FLUME\_HOME/bin

（3）修改配置文件

conf/flume-env.sh 文件中配置JAVA\_HOME

conf目录下新建一个.conf文件，如myflume.conf

# 声明三种组件
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    # 定义source信息
    a1.sources.r1.type = netcat
    a1.sources.r1.bind = localhost
    a1.sources.r1.port = 8888
    
    # 定义sink信息
    a1.sinks.k1.type = logger
    
    # 定义channel信息
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    # 绑定在一起
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

启动flume：

flume-ng agent -n a1 -c conf -f myflume.conf

测试：

在当前节点上执行：nc localhost 8888

![20180507231319196][]

![20180507231302312][]

[20180507230026472]: /images/20220525/ab59513161d04b04ad343aa563096c95.png
[Link 1]: http://flume.apache.org/FlumeDeveloperGuide.html#sink
[20180507230418199]: /images/20220525/ff83daa9463f49d78558716e0b511673.png
[20180507230458229]: /images/20220525/763516ca4e894adbae9c8f4cae568f54.png
[20180507230509281]: /images/20220525/e3a07c46097d41b5a936decf2c70e58d.png
[SouthEast]: /images/20220525/60cdcf098d7446f1993fb9d78349eb3a.png
[20180507231319196]: /images/20220525/006bb03cf2014ff0a6eb4c8e0f4c0294.png
[20180507231302312]: /images/20220525/58c2c9bb3c2d4f0fb346d0649534c700.png