Flume 介绍

概述

Flume 是一个高可用的，高可靠的，分布式的海量日志采集、聚合和传输的系统。 Flume 支持定制各类数据发送方，用于收集各类型数据；Flume 支持定制各类数据发送方，用于收集各类型数据；同时， Flume 提供对数据进行简单处理，并写到各种数据接受方（可定制）的能力。 一般的采集需求，通过对 flume 的简单配置即可实现。 针对特殊场景也具备良好的自定义扩展能力。 因此， flume 可以适用于大部分的日常数据采集场景。
当前 Flume 有两个版本。 Flume 0.9X 版本的统称 Flume OG（ originalgeneration）， Flume1.X 版本的统称 Flume NG（next generation）。由于 FlumeNG 经过核心组件、核心配置以及代码架构重构，与 Flume OG 有很大不同，使用时请注意区分。 改动的另一原因是将 Flume 纳入 apache 旗下， cloudera Flume改名为 Apache Flume。
这里使用的是Flume1.X版本

运行机制

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

核心的角色是 agent， agent 本身是一个 Java 进程， 一般运行在日志收集节点。 flume 采集系统就是由一个个 agent 所连接起来形成。

agent有三个组件:

• Source：采集源，用于跟数据源对接，以获取数据；
• Channel： agent 内部的数据传输通道；
• Sink：下沉地，采集数据的传送目的，用于往下一级 agent 传递数据或者往最终存储系统传递数据；

在数据的传输的过程中，流动的是 event， 它是 Flume 内部数据传输的最基本单元
一个完整的 event 包括： event headers、 event body、 event 信息， 其中event 信息就是 flume 收集到的日记记录。

Flume的结构

单个 agent 采集数据

多级 agent 串联

Flume 安装

上传安装包并解压到要安装的目录
修改conf 目录下的 flume-env.sh 文件, 配置好 JAVA_HOME 即可

Flume 使用

Flume 的使用主要是根据数据采集需求配置采集方案
在相应的节点上启动 flume agent时，指定采集方案配置文件

采集目录到 HDFS

根据需求，首先定义以下 3 大要素

• 采集源，即 source——监控文件目录 : spooldir
• 下沉目标，即 sink——HDFS 文件系统 : hdfs sink
• source 和 sink 之间的传递通道——channel，可用 file channel 也可以用内存 channel

例: 采集/root/logs目录到hdfs的/flume/events/%y-%m-%d/%H%M/路径

# 组件别名
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# 数据源描述
# 注意：不能往监控目中重复丢同名文件
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /root/logs
a1.sources.r1.fileHeader = true

# 下沉地描述
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /flume/events/%y-%m-%d/%H%M/
a1.sinks.k1.hdfs.filePrefix = events-
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute
a1.sinks.k1.hdfs.rollInterval = 3
a1.sinks.k1.hdfs.rollSize = 20
a1.sinks.k1.hdfs.rollCount = 5
a1.sinks.k1.hdfs.batchSize = 1
a1.sinks.k1.hdfs.useLocalTimeStamp = true
#生成的文件类型，默认是 Sequencefile，可用 DataStream，则为普通文本
a1.sinks.k1.hdfs.fileType = DataStream

# 管道描述
a1.channels.c1.type = memory
# capacity：默认该通道中最大的可以存储的 event 数量
a1.channels.c1.capacity = 1000
# trasactionCapacity：每次最大可以从 source 中拿到或者送到 sink 中的 event数量
a1.channels.c1.transactionCapacity = 100

# 绑定组件
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

把此文件放在flume根路径下myconf文件夹内命名为myconf/spooldir_source_hdfs_sink.conf

启动任务:

bin/flume-ng agent -c conf -f myconf/spooldir_source_hdfs_sink.conf -n a1 -Dflume.root.logger=INFO,console

采集文件到 HDFS

根据需求，首先定义以下 3 大要素

• 采集源，即 source——监控文件内容更新 : exec ‘tail -F file’
• 下沉目标，即 sink——HDFS 文件系统 : hdfs sink
• Source 和 sink 之间的传递通道——channel，可用 file channel 也可以用内存 channel

例:

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /root/logs/test.log
a1.sources.r1.channels = c1
# Describe the sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /flume/tailout/%y-%m-%d/%H%M/
a1.sinks.k1.hdfs.filePrefix = eventsa1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute
a1.sinks.k1.hdfs.rollInterval = 3
a1.sinks.k1.hdfs.rollSize = 20
a1.sinks.k1.hdfs.rollCount = 5
a1.sinks.k1.hdfs.batchSize = 1
a1.sinks.k1.hdfs.useLocalTimeStamp = true
#生成的文件类型，默认是 Sequencefile，可用 DataStream，则为普通文本
a1.sinks.k1.hdfs.fileType = DataStream
# 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

其中的相关参数:

• rollInterval
  hdfs sink 间隔多长将临时文件滚动成最终目标文件，默认值： 30 单位：秒；
  如果设置成 0，则表示不根据时间来滚动文件；
  注：滚动（roll）指的是， hdfs sink 将临时文件重命名成最终目标文件，并新打开一个临时文件来写入数据；
• rollSize
  当临时文件达到该大小（默认值：1024 单位： bytes）时，滚动成目标文件；如果设置成 0，则表示不根据临时文件大小来滚动文件
• rollCount
  当 events 数据达到该数量 (默认值：10) 时候，将临时文件滚动成目标文件；如果设置成 0，则表示不根据 events 数据来滚动文件
• round
  默认值：false
  是否启用时间上的“舍弃” ，这里的“舍弃” ，类似于“四舍五入”
• roundValue
  时间上进行“舍弃” 的值(默认值： 1)
• roundUnit
  时间上进行“舍弃” 的单位，包含： second,minute,hour (默认值： seconds)

启动任务:

bin/flume-ng agent -c conf -f myconf/exec_source_hdfs_sink.conf -n a1 -Dflume.root.logger=INFO,console

Flume 的 load-balance、failover

• load-balance 负载均衡机制
  负载均衡是用于解决一台机器(一个进程)无法解决所有请求而产生的一种算法
  Load balancing Sink Processor 能够实现 load balance 功能, 如下图所示
  
  Agent1节点通过多个Sink将event均衡到多个独立的Agent
  Sink实例配置如下:

  a1.sinkgroups = g1
  a1.sinkgroups.g1.sinks = k1 k2 k3
  a1.sinkgroups.g1.processor.type = load_balance
  # 如果开启，则将失败的 sink 放入黑名单
  a1.sinkgroups.g1.processor.backoff = true
  # 另外还支持 random
  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将会被降级到一个池，在这些池中它们被分配一个冷却时间，随着故障的连续， 在重试之前冷却时间增加。 一旦 Sink 成功发送一个事件，它将恢复到活动池。 Sink 具有与之相关的优先级，数量越大，优先级越高
  如果没有指定优先级，则根据在配置中指定 Sink 的顺序来确定优先级
  示例配置:

  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
  # 失败的 Sink 的最大回退期（millis）
  a1.sinkgroups.g1.processor.maxpenalty = 20000