flume基础（一）

柔情只为你懂 2022-05-21 02:46 153阅读 0赞

官网：http://flume.apache.org/FlumeUserGuide.html

概念：**flume是一个分布式、可靠、和高可用的海量日志采集、聚合和传输的系统**。支持在日志系统中定制各类数据发送方，用于收集数据; 同时，Flume提供对数据进行简单处理，并写到各种数据接受方(比如文本、HDFS、Hbase等)的能力 。

# 1、Flume 的一些核心概念： #

<table> 
 <tbody> 
  <tr> 
   <th style="border-color:#cccccc;vertical-align:top;">组件</th> 
   <th style="border-color:#cccccc;vertical-align:top;">功能</th> 
  </tr> 
  <tr> 
   <td style="border-color:#cccccc;vertical-align:top;">Agent</td> 
   <td style="border-color:#cccccc;vertical-align:top;">使用JVM 运行Flume。每台机器运行一个agent，但是可以在一个agent中包含多个sources和sinks。</td> 
  </tr> 
  <tr> 
   <td style="border-color:#cccccc;vertical-align:top;">Client</td> 
   <td style="border-color:#cccccc;vertical-align:top;">生产数据，运行在一个独立的线程。</td> 
  </tr> 
  <tr> 
   <td style="border-color:#cccccc;vertical-align:top;">Source</td> 
   <td style="border-color:#cccccc;vertical-align:top;">从Client收集数据，传递给Channel。</td> 
  </tr> 
  <tr> 
   <td style="border-color:#cccccc;vertical-align:top;">Sink</td> 
   <td style="border-color:#cccccc;vertical-align:top;">从Channel收集数据，运行在一个独立线程。</td> 
  </tr> 
  <tr> 
   <td style="border-color:#cccccc;vertical-align:top;">Channel</td> 
   <td style="border-color:#cccccc;vertical-align:top;">连接 sources 和 sinks ，这个有点像一个队列。</td> 
  </tr> 
  <tr> 
   <td style="border-color:#cccccc;vertical-align:top;">Events</td> 
   <td style="border-color:#cccccc;vertical-align:top;">可以是日志记录、 avro 对象等。</td> 
  </tr> 
 </tbody> 
</table>

## 1.1 数据流模型 ##

Flume以agent为最小的独立运行单位。一个agent就是一个JVM。单agent由Source、Sink和Channel三大组件构成，如下图：

![Agent component diagram][]  图一

Flume的数据流由事件(Event)贯穿始终。事件是Flume的基本数据单位，它携带日志数据(字节数组形式)并且携带有头信息，这些Event由Agent外部的Source，比如上图中的Web Server生成。当Source捕获事件后会进行特定的格式化，然后Source会把事件推入(单个或多个)Channel中。你可以把Channel看作是一个缓冲区，它将保存事件直到Sink处理完该事件。Sink负责持久化日志或者把事件推向另一个Source。  
很直白的设计，其中值得注意的是，Flume提供了大量内置的Source、Channel和Sink类型。不同类型的Source,Channel和Sink可以自由组合。组合方式基于用户设置的配置文件，非常灵活。比如：Channel可以把事件暂存在内存里，也可以持久化到本地硬盘上。Sink可以把日志写入HDFS, HBase，甚至是另外一个Source等等。  
如果你以为Flume就这些能耐那就大错特错了。Flume支持用户建立多级流，也就是说，多个agent可以协同工作，并且支持Fan-in、Fan-out、Contextual Routing、Backup Routes。如下图所示：

![A fan-out flow using a (multiplexing) channel selector][A fan-out flow using a _multiplexing_ channel selector]

## 1.2 高可靠性 ##

作为生产环境运行的软件，高可靠性是必须的。  
从单agent来看，Flume使用基于事务的数据传递方式来保证事件传递的可靠性。Source和Sink被封装进一个事务。事件被存放在Channel中直到该事件被处理，Channel中的事件才会被移除。这是Flume提供的点到点的可靠机制。  
从多级流来看，前一个agent的sink和后一个agent的source同样有它们的事务来保障数据的可靠性。

## 1.3 可恢复性 ##

还是靠Channel。推荐使用FileChannel，事件持久化在本地文件系统里(性能较差)。

# 2、Flume 整体架构介绍 #

Flume架构整体上看就是 source-->channel-->sink 的三层架构（参见最上面的 图一），类似生成者和消费者的架构，他们之间通过queue（channel）传输，解耦。

Source:完成对日志数据的收集，分成 transtion 和 event 打入到channel之中。

Channel:主要提供一个队列的功能，对source提供中的数据进行简单的缓存。

Sink:取出Channel中的数据，进行相应的存储文件系统，数据库，或者提交到远程服务器。

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

对于直接读取文件Source, 主要有两种方式：

## 2.1 Exec source ##

可通过写Unix command的方式组织数据，最常用的就是tail -F \[file\]。  
可以实现实时传输，但在flume不运行和脚本错误时，会丢数据，也不支持断点续传功能。因为没有记录上次文件读到的位置，从而没办法知道，下次再读时，从什么地方开始读。特别是在日志文件一直在增加的时候。flume的source挂了。等flume的source再次开启的这段时间内，增加的日志内容，就没办法被source读取到了。不过flume有一个execStream的扩展，可以自己写一个监控日志增加情况，把增加的日志，通过自己写的工具把增加的内容，传送给flume的node。再传送给sink的node。要是能在tail类的source中能支持，在node挂掉这段时间的内容，等下次node开启后在继续传送，那就更完美了。

## 2.2 Spooling Directory Source ##

SpoolSource:是监测配置的目录下新增的文件，并将文件中的数据读取出来，可实现准实时。需要注意两点：1、拷贝到spool目录下的文件不可以再打开编辑。2、spool目录下不可包含相应的子目录。在实际使用的过程中，可以结合log4j使用，使用log4j的时候，将log4j的文件分割机制设为1分钟一次，将文件拷贝到spool的监控目录。log4j有一个TimeRolling的插件，可以把log4j分割的文件到spool目录。基本实现了实时的监控。Flume在传完文件之后，将会修改文件的后缀，变为.COMPLETED（后缀也可以在配置文件中灵活指定）   
ExecSource，SpoolSource对比：ExecSource可以实现对日志的实时收集，但是存在Flume不运行或者指令执行出错时，将无法收集到日志数据，无法何证日志数据的完整性。SpoolSource虽然无法实现实时的收集数据，但是可以使用以分钟的方式分割文件，趋近于实时。如果应用无法实现以分钟切割日志文件的话，可以两种收集方式结合使用。   
Channel有多种方式：有MemoryChannel, JDBC Channel, MemoryRecoverChannel, FileChannel。MemoryChannel可以实现高速的吞吐，但是无法保证数据的完整性。MemoryRecoverChannel在官方文档的建议上已经建义使用FileChannel来替换。FileChannel保证数据的完整性与一致性。在具体配置FileChannel时，建议FileChannel设置的目录和程序日志文件保存的目录设成不同的磁盘，以便提高效率。   
Sink在设置存储数据时，可以向文件系统中，数据库中，hadoop中储数据，在日志数据较少时，可以将数据存储在文件系中，并且设定一定的时间间隔保存数据。在日志数据较多时，可以将相应的日志数据存储到Hadoop中，便于日后进行相应的数据分析。

# 3、常用架构、功能配置示例 #

## 3.1 先来个简单的：单节点 Flume 配置 ##

# example.conf: A single-node Flume configuration
     # Name the components on this agent
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
     # Describe/configure the source
    a1.sources.r1.type = netcat
    a1.sources.r1.bind = localhost
    a1.sources.r1.port = 44444
     # Describe the sink
    a1.sinks.k1.type = logger
     # Use a channel which buffers events in memory
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
     # Bind the source and sink to the channel
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

将上述配置存为：example.conf

然后我们就可以启动 Flume 了：

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

PS：-Dflume.root.logger=INFO,console 仅为 debug 使用，请勿生产环境生搬硬套，否则大量的日志会返回到终端。。。

\-c/--conf 后跟配置目录，-f/\--conf-file 后跟具体的配置文件，-n/\--name 指定agent的名称

然后我们再开一个 shell 终端窗口，telnet 上配置中侦听的端口，就可以发消息看到效果了：

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

Flume 终端窗口此时会打印出如下信息，就表示成功了：

12/06/19 15:32:19 INFO source.NetcatSource: Source starting
    12/06/19 15:32:19 INFO source.NetcatSource: Created serverSocket:sun.nio.ch.ServerSocketChannelImpl[/127.0.0.1:44444]
    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 Agent 算是部署成功了！

## 3.2 单节点 Flume 直接写入 HDFS ##

# Define a memory channel called ch1 on agent1
    agent1.channels.ch1.type = memory
    agent1.channels.ch1.capacity = 100000
    agent1.channels.ch1.transactionCapacity = 100000
    agent1.channels.ch1.keep-alive = 30
     # Define an Avro source called avro-source1 on agent1 and tell it
    # to bind to 0.0.0.0:41414. Connect it to channel ch1.
    #agent1.sources.avro-source1.channels = ch1
    #agent1.sources.avro-source1.type = avro
    #agent1.sources.avro-source1.bind = 0.0.0.0
    #agent1.sources.avro-source1.port = 41414
    #agent1.sources.avro-source1.threads = 5
     #define source monitor a file
    agent1.sources.avro-source1.type = exec
    agent1.sources.avro-source1.shell = /bin/bash -c
    agent1.sources.avro-source1.command = tail -n +0 -F /home/storm/tmp/id.txt
    agent1.sources.avro-source1.channels = ch1
    agent1.sources.avro-source1.threads = 5
     # Define a logger sink that simply logs all events it receives
    # and connect it to the other end of the same channel.
    agent1.sinks.log-sink1.channel = ch1
    agent1.sinks.log-sink1.type = hdfs
    agent1.sinks.log-sink1.hdfs.path = hdfs://192.168.1.111:8020/flumeTest
    agent1.sinks.log-sink1.hdfs.writeFormat = Text
    agent1.sinks.log-sink1.hdfs.fileType = DataStream
    agent1.sinks.log-sink1.hdfs.rollInterval = 0
    agent1.sinks.log-sink1.hdfs.rollSize = 1000000
    agent1.sinks.log-sink1.hdfs.rollCount = 0
    agent1.sinks.log-sink1.hdfs.batchSize = 1000
    agent1.sinks.log-sink1.hdfs.txnEventMax = 1000
    agent1.sinks.log-sink1.hdfs.callTimeout = 60000
    agent1.sinks.log-sink1.hdfs.appendTimeout = 60000
     # Finally, now that we've defined all of our components, tell
    # agent1 which ones we want to activate.
    agent1.channels = ch1
    agent1.sources = avro-source1
    agent1.sinks = log-sink1

启动如下命令，就可以在 hdfs 上看到效果了。

../bin/flume-ng agent --conf ../conf/ -f flume\_directHDFS.conf -n agent1 -Dflume.root.logger=INFO,console

PS：实际环境中有这样的需求，通过在多个agent端tail日志，发送给collector，collector再把数据收集，统一发送给HDFS存储起来，当HDFS文件大小超过一定的大小或者超过在规定的时间间隔会生成一个文件。  
Flume 实现了两个Trigger，分别为SizeTriger（在调用HDFS输出流写的同时，count该流已经写入的大小总和，若超过一定大小，则创建新的文件和输出流，写入操作指向新的输出流，同时close以前的输出流）和TimeTriger（开启定时器，当到达该点时，自动创建新的文件和输出流，新的写入重定向到该流中，同时close以前的输出流）。

## 3.3 来一个常见架构：多 agent 汇聚写入 HDFS ##

![A fan-in flow using Avro RPC to consolidate events in one place][]

### 3.3.1 在各个webserv日志机上配置 Flume Client ###

# clientMainAgent
    clientMainAgent.channels = c1
    clientMainAgent.sources  = s1
    clientMainAgent.sinks    = k1 k2
    # clientMainAgent sinks group
    clientMainAgent.sinkgroups = g1
    # clientMainAgent Spooling Directory Source
    clientMainAgent.sources.s1.type = spooldir
    clientMainAgent.sources.s1.spoolDir  =/dsap/rawdata/
    clientMainAgent.sources.s1.fileHeader = true
    clientMainAgent.sources.s1.deletePolicy =immediate
    clientMainAgent.sources.s1.batchSize =1000
    clientMainAgent.sources.s1.channels =c1
    clientMainAgent.sources.s1.deserializer.maxLineLength =1048576
    # clientMainAgent FileChannel
    clientMainAgent.channels.c1.type = file
    clientMainAgent.channels.c1.checkpointDir = /var/flume/fchannel/spool/checkpoint
    clientMainAgent.channels.c1.dataDirs = /var/flume/fchannel/spool/data
    clientMainAgent.channels.c1.capacity = 200000000
    clientMainAgent.channels.c1.keep-alive = 30
    clientMainAgent.channels.c1.write-timeout = 30
    clientMainAgent.channels.c1.checkpoint-timeout=600
    # clientMainAgent Sinks
    # k1 sink
    clientMainAgent.sinks.k1.channel = c1
    clientMainAgent.sinks.k1.type = avro
    # connect to CollectorMainAgent
    clientMainAgent.sinks.k1.hostname = flume115
    clientMainAgent.sinks.k1.port = 41415 
    # k2 sink
    clientMainAgent.sinks.k2.channel = c1
    clientMainAgent.sinks.k2.type = avro
    # connect to CollectorBackupAgent
    clientMainAgent.sinks.k2.hostname = flume116
    clientMainAgent.sinks.k2.port = 41415
    # clientMainAgent sinks group
    clientMainAgent.sinkgroups.g1.sinks = k1 k2
    # load_balance type
    clientMainAgent.sinkgroups.g1.processor.type = load_balance
    clientMainAgent.sinkgroups.g1.processor.backoff   = true
    clientMainAgent.sinkgroups.g1.processor.selector  = random

../bin/flume-ng agent --conf ../conf/ -f flume\_Consolidation.conf -n clientMainAgent -Dflume.root.logger=DEBUG,console

### 3.3.2 在汇聚节点配置 Flume server ###

# collectorMainAgent
    collectorMainAgent.channels = c2
    collectorMainAgent.sources  = s2
    collectorMainAgent.sinks    =k1 k2
    # collectorMainAgent AvroSource
    #
    collectorMainAgent.sources.s2.type = avro
    collectorMainAgent.sources.s2.bind = flume115
    collectorMainAgent.sources.s2.port = 41415
    collectorMainAgent.sources.s2.channels = c2
     # collectorMainAgent FileChannel
    #
    collectorMainAgent.channels.c2.type = file
    collectorMainAgent.channels.c2.checkpointDir =/opt/var/flume/fchannel/spool/checkpoint
    collectorMainAgent.channels.c2.dataDirs = /opt/var/flume/fchannel/spool/data,/work/flume/fchannel/spool/data
    collectorMainAgent.channels.c2.capacity = 200000000
    collectorMainAgent.channels.c2.transactionCapacity=6000
    collectorMainAgent.channels.c2.checkpointInterval=60000
    # collectorMainAgent hdfsSink
    collectorMainAgent.sinks.k2.type = hdfs
    collectorMainAgent.sinks.k2.channel = c2
    collectorMainAgent.sinks.k2.hdfs.path = hdfs://db-cdh-cluster/flume%{dir}
    collectorMainAgent.sinks.k2.hdfs.filePrefix =k2_%{file}
    collectorMainAgent.sinks.k2.hdfs.inUsePrefix =_
    collectorMainAgent.sinks.k2.hdfs.inUseSuffix =.tmp
    collectorMainAgent.sinks.k2.hdfs.rollSize = 0
    collectorMainAgent.sinks.k2.hdfs.rollCount = 0
    collectorMainAgent.sinks.k2.hdfs.rollInterval = 240
    collectorMainAgent.sinks.k2.hdfs.writeFormat = Text
    collectorMainAgent.sinks.k2.hdfs.fileType = DataStream
    collectorMainAgent.sinks.k2.hdfs.batchSize = 6000
    collectorMainAgent.sinks.k2.hdfs.callTimeout = 60000
    collectorMainAgent.sinks.k1.type = hdfs
    collectorMainAgent.sinks.k1.channel = c2
    collectorMainAgent.sinks.k1.hdfs.path = hdfs://db-cdh-cluster/flume%{dir}
    collectorMainAgent.sinks.k1.hdfs.filePrefix =k1_%{file}
    collectorMainAgent.sinks.k1.hdfs.inUsePrefix =_
    collectorMainAgent.sinks.k1.hdfs.inUseSuffix =.tmp
    collectorMainAgent.sinks.k1.hdfs.rollSize = 0
    collectorMainAgent.sinks.k1.hdfs.rollCount = 0
    collectorMainAgent.sinks.k1.hdfs.rollInterval = 240
    collectorMainAgent.sinks.k1.hdfs.writeFormat = Text
    collectorMainAgent.sinks.k1.hdfs.fileType = DataStream
    collectorMainAgent.sinks.k1.hdfs.batchSize = 6000
    collectorMainAgent.sinks.k1.hdfs.callTimeout = 60000

../bin/flume-ng agent --conf ../conf/ -f flume\_Consolidation.conf -n collectorMainAgent -Dflume.root.logger=DEBUG,console

上面采用的就是类似 cs 架构，各个 flume agent 节点先将各台机器的日志汇总到 Consolidation 节点，然后再由这些节点统一写入 HDFS，并且采用了负载均衡的方式，你还可以配置高可用的模式等等。

总结：

![70][]

转载：https://my.oschina.net/leejun2005/blog/288136 （赞）

参考资料：

https://www.jb51.net/article/53542.htm

https://blog.csdn.net/xiaochawan/article/details/8986489

[Agent component diagram]: /images/20220521/2256b899d25f48ad81b8a7f85ce975da.png
[A fan-out flow using a _multiplexing_ channel selector]: /images/20220521/059dfbf490454383b59a2d68181e9fbe.png
[A fan-in flow using Avro RPC to consolidate events in one place]: /images/20220521/460c1cd9881b4dc38ae972726efe00d1.png
[70]: /images/20220521/d8bf5fa065984b98834882577a71cdd3.png