2.2.4 实时监控目录下的多个追加文件

Exec source 适用于监控一个实时追加的文件，不能实现断点续传；

Spooldir Source适合用于同步新文件，但不适合对实时追加日志的文件进行监听并同步；

而 Taildir Source适合用于监听多个实时追加的文件，并且能够实现断点续传。

1）案例需求:使用 Flume 监听整个目录的实时追加文件，并上传至 HDFS

2）需求分析:

3）实现步骤：

（1）创建配置文件 flume-taildir-hdfs.conf

创建一个文件

[atguigu@hadoop102 job]$ vim flume-taildir-hdfs.conf

添加如下内容

a3.sources = r3
a3.sinks = k3
a3.channels = c3
# Describe/configure the source
a3.sources.r3.type = TAILDIR
a3.sources.r3.positionFile = /opt/module/flume/tail_dir.json
a3.sources.r3.filegroups = f1 f2
a3.sources.r3.filegroups.f1 = /opt/module/flume/files/.*file.*
a3.sources.r3.filegroups.f2 = /opt/module/flume/files2/.*log.*
# Describe the sink
a3.sinks.k3.type = hdfs
a3.sinks.k3.hdfs.path = hdfs://hadoop102:9820/flume/upload2/%Y%m%d/%H
#上传文件的前缀
a3.sinks.k3.hdfs.filePrefix = upload-
#是否按照时间滚动文件夹
a3.sinks.k3.hdfs.round = true
#多少时间单位创建一个新的文件夹
a3.sinks.k3.hdfs.roundValue = 1
#重新定义时间单位
a3.sinks.k3.hdfs.roundUnit = hour
#是否使用本地时间戳
a3.sinks.k3.hdfs.useLocalTimeStamp = true
#积攒多少个 Event 才 flush 到 HDFS 一次
a3.sinks.k3.hdfs.batchSize = 100
#设置文件类型，可支持压缩
a3.sinks.k3.hdfs.fileType = DataStream
#多久生成一个新的文件
a3.sinks.k3.hdfs.rollInterval = 60
#设置每个文件的滚动大小大概是 128M
a3.sinks.k3.hdfs.rollSize = 134217700
#文件的滚动与 Event 数量无关
a3.sinks.k3.hdfs.rollCount = 0
# Use a channel which buffers events in memory
a3.channels.c3.type = memory
a3.channels.c3.capacity = 1000
a3.channels.c3.transactionCapacity = 100
# Bind the source and sink to the channel
a3.sources.r3.channels = c3
a3.sinks.k3.channel = c3

骚戴理解：

a3.sources.r3.type是apache flume中用于指定数据源（source）的类型的属性。以下是a3.sources.r3.type可能的值列表：

avro：avrosource，从avro客户端接收事件

netcat：netcatsource，从tcp/ip套接字接收事件

exec：execsource，运行进程并读取其stdout作为事件

spooldir：spooldirectorysource，监视包含事件文件的目录，并将文件内容发送到通道

syslog：syslogtcpsource或syslogudpsource，从tcp或udp端口接收syslog事件

thrift：thriftsource，从thrift客户端接收事件

http：httpsource，从http客户端接收事件

jms：jms source，监听jms队列以接收消息

kafka：kafka source，从kafka topic接收消息

a3.sources.r3.positionFile = /opt/module/flume/tail_dir.json的意思是设置断点信息的存储路径，这里面记录着各种实时同步的信息，如果异常或者宕机导致同步失败了就可以从这里记录的断点继续同步，而不需要重头开始！

以下三句的意思是定义两个文件组，目的就是用了监控多个目录中的文件，f1就监控files里文件名中包含file的文件，f2就监控files里文件名中包含log的文件，如果文件有变化，追加了信息，那就同步到hdfs

a3.sources.r3.filegroups = f1 f2

a3.sources.r3.filegroups.f1 = /opt/module/flume/files/.*file.*

a3.sources.r3.filegroups.f2 = /opt/module/flume/files2/.*log.*

（2）启动监控文件夹命令

1. [atguigu@hadoop102 flume]$ bin/flume-ng agent --conf conf/ --name 
2. a3 --conf-file job/flume-taildir-hdfs.conf

（3）向 files 文件夹中追加内容

在/opt/module/flume 目录下创建 files 文件夹

[atguigu@hadoop102 flume]$ mkdir files

向 upload 文件夹中添加文件

[atguigu@hadoop102 files]$ echo hello >> file1.txt
[atguigu@hadoop102 files]$ echo atguigu >> file2.txt

4）查看 HDFS 上的数据

Taildir 说明：

Taildir Source 维护了一个 json 格式的 position File，其会定期的往 position File中更新每个文件读取到的最新的位置，因此能够实现断点续传。

Position File 的格式如下：

{"inode":2496272,"pos":12,"file":"/opt/module/flume/files/file1.t
xt"}
{"inode":2496275,"pos":12,"file":"/opt/module/flume/files/file2.t
xt"}

第 3 章 Flume 进阶

3.1 Flume 事务

3.2 Flume Agent 内部原理

重要组件：

1）ChannelSelector

ChannelSelector 的作用就是选出 Event 将要被发往哪个 Channel。其共有两种类型，分别是Replicating（复制）和 Multiplexing（多路复用）。

Replicating会将同一个 Event 发往所有的 Channel，Multiplexing 会根据相应的原则，将不同的 Event 发往不同的 Channel。

骚戴理解：Replicating的意思是分发Event给每个Channel，在实际的开发中，一台服务器产生的日志类型可能有很多种，不同类型的日志可能需要发送到不同的分析系统。此时会用到 Flume 拓扑结构中的 Multiplexing 结构，Multiplexing的原理是，根据 event 中 Header 的某个 key 的值，将不同的 event 发送到不同的 Channel 中，所以我们需要自定义一个 Interceptor，为不同类型的 event 的 Header 中的 key 赋予不同的值。

2）SinkProcessor

SinkProcessor 共 有 三 种 类 型 ， 分 别 是 DefaultSinkProcessor 、LoadBalancingSinkProcessor 和 FailoverSinkProcessorDefaultSinkProcessor 对 应 的 是 单 个 的 Sink ，LoadBalancingSinkProcessor 和FailoverSinkProcessor 对应的是 Sink Group（Sink组），LoadBalancingSinkProcessor 可以实现负载均衡的功能，FailoverSinkProcessor 可以错误恢复的功能。

骚戴理解：首先这里要知道一个Sink只能绑定一个Channel，但是一个Channel可以绑定多个Sink！

failoversinkprocessor 是一个用于 apache flume 的 sink 处理器，用于处理事件流并将其写入外部存储系统（如 hadoop、hbase 或 elasticsearch）。它被称为“故障切换”处理器，因为它具有在主要目标无法使用时自动切换到备用目标的能力。它需要给多个Sink定义一个优先级，值越大优先级越高，假如有三个Sink，设置优先级为100,50,10，然后假如优先级为100的Sink挂掉了，那么就会换到优先级50的Sink

通常情况下，flume 将事件传输到单个目标。但是，在某些情况下，这种方法可能导致数据丢失或停机时间增加。例如，如果目标系统不可用，则无法传输事件。这就是 failoversinkprocessor 可以派上用场的地方。failoversinkprocessor 允许您定义多个目标，如果主要目标失败，则会自动切换到备用目标。当主要目标恢复正常时，可以再次切换回主要目标。这确保了高可用性和数据完整性

3.3 Flume 拓扑结构

3.3.1 简单串联

这种模式是将多个 flume 顺序连接起来了，从最初的 source 开始到最终 sink 传送的目的存储系统。此模式不建议桥接过多的 flume 数量， flume 数量过多不仅会影响传输速率，而且一旦传输过程中某个节点 flume 宕机，会影响整个传输系统。

骚戴理解：avro 是一种数据序列化系统，用于在不同的应用程序、编程语言和平台之间传输数据。它是由 apache 软件基金会开发的一种开放源代码项目，并且被广泛用于 hadoop 生态系统中，如使用 hadoop 的数据处理工具 hive 和 pig。

avro 通过使用二进制编码对数据进行序列化，因此比其他序列化系统（如 xml 或 json）更高效，更节省空间。此外，avro 模式支持动态生成，这意味着您可以定义数据结构并为其生成模式，而无需预先编译代码。另一个 avro 的优势在于其跨语言支持。avro 支持多种编程语言，包括 c、c++、python、ruby、java 和 javascript 等。这使得在不同的应用程序之间共享数据变得容易和灵活。

总之，avro 是一种用于跨平台、跨语言数据序列化的系统，具有高效性和灵活性等优点。

3.3.2 复制和多路复用

Flume 支持将事件流向一个或者多个目的地。这种模式可以将相同数据复制到多个channel 中，或者将不同数据分发到不同的 channel 中，sink 可以选择传送到不同的目的地

3.3.3 负载均衡和故障转移

Flume支持使用将多个sink逻辑上分到一个sink组，sink组配合不同的SinkProcessor可以实现负载均衡和错误恢复的功能。

3.3.4 聚合

聚合这种模式是我们最常见的，也非常实用，日常 web 应用通常分布在上百个服务器，大者甚至上千个、上万个服务器。产生的日志，处理起来也非常麻烦。用 flume 的这种组合方式能很好的解决这一问题，每台服务器部署一个 flume 采集日志，传送到一个集中收集日志的 flume，再由此 flume 上传到 hdfs、hive、hbase 等，进行日志分析。

3.4 Flume 开发案例

3.4.1 复制和多路复用

1）案例需求

使用 Flume-1 监控文件变动，Flume-1 将变动内容传递给 Flume-2，Flume-2 负责存储到 HDFS。同时 Flume-1 将变动内容传递给 Flume-3，Flume-3 负责输出到 Local FileSystem。

2）需求分析：

单数据源多出口案例（选择器）

3）实现步骤：

（1）准备工作

在/opt/module/flume/job 目录下创建 group1 文件夹

[atguigu@hadoop102 job]$ cd group1/

在/opt/module/datas/目录下创建 flume3 文件夹

[atguigu@hadoop102 datas]$ mkdir flume3

（2）创建 flume-file-flume.conf

配置 1 个接收日志文件的 source 和两个 channel、两个 sink，分别输送给 flume-flume-hdfs 和 flume-flume-dir。

编辑flume1的配置文件

[atguigu@hadoop102 group1]$ vim flume-file-flume.conf

添加如下内容

# Name the components on this agent
a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1 c2
# 将数据流复制给所有 channel
a1.sources.r1.selector.type = replicating
# Describe/configure the source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /opt/module/hive/logs/hive.log
a1.sources.r1.shell = /bin/bash -c
# Describe the sink
# sink 端的 avro 是一个数据发送者
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = hadoop102
a1.sinks.k1.port = 4141
a1.sinks.k2.type = avro
a1.sinks.k2.hostname = hadoop102
a1.sinks.k2.port = 4142
# Describe the channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
a1.channels.c2.type = memory
a1.channels.c2.capacity = 1000
a1.channels.c2.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1 c2
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c2

（3）创建 flume-flume-hdfs.conf

配置上级 Flume 输出的 Source，输出是到 HDFS 的 Sink。

编辑flume2的配置文件

[atguigu@hadoop102 group1]$ vim flume-flume-hdfs.conf

添加如下内容
# Name the components on this agent
a2.sources = r1
a2.sinks = k1
a2.channels = c1
# Describe/configure the source
# source 端的 avro 是一个数据接收服务
a2.sources.r1.type = avro
a2.sources.r1.bind = hadoop102
a2.sources.r1.port = 4141
# Describe the sink
a2.sinks.k1.type = hdfs
a2.sinks.k1.hdfs.path = hdfs://hadoop102:9820/flume2/%Y%m%d/%H  
#上传文件的前缀
a2.sinks.k1.hdfs.filePrefix = flume2-
#是否按照时间滚动文件夹
a2.sinks.k1.hdfs.round = true
#多少时间单位创建一个新的文件夹
a2.sinks.k1.hdfs.roundValue = 1
#重新定义时间单位
a2.sinks.k1.hdfs.roundUnit = hour
#是否使用本地时间戳
a2.sinks.k1.hdfs.useLocalTimeStamp = true
#积攒多少个 Event 才 flush 到 HDFS 一次
a2.sinks.k1.hdfs.batchSize = 100
#设置文件类型，可支持压缩
a2.sinks.k1.hdfs.fileType = DataStream
#多久生成一个新的文件
a2.sinks.k1.hdfs.rollInterval = 30
#设置每个文件的滚动大小大概是 128M
a2.sinks.k1.hdfs.rollSize = 134217700
#文件的滚动与 Event 数量无关
a2.sinks.k1.hdfs.rollCount = 0
# Describe the channel
a2.channels.c1.type = memory
a2.channels.c1.capacity = 1000
a2.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a2.sources.r1.channels = c1
a2.sinks.k1.channel = c1

（4）创建 flume-flume-dir.conf

配置上级 Flume 输出的 Source，输出是到本地目录的 Sink。

编辑flume3的配置文件

[atguigu@hadoop102 group1]$ vim flume-flume-dir.conf

添加如下内容

# Name the components on this agent
a3.sources = r1
a3.sinks = k1
a3.channels = c2
# Describe/configure the source
a3.sources.r1.type = avro
a3.sources.r1.bind = hadoop102
a3.sources.r1.port = 4142
# Describe the sink
a3.sinks.k1.type = file_roll
a3.sinks.k1.sink.directory = /opt/module/data/flume3
# Describe the channel
a3.channels.c2.type = memory
a3.channels.c2.capacity = 1000
a3.channels.c2.transactionCapacity = 100  
# Bind the source and sink to the channel
a3.sources.r1.channels = c2
a3.sinks.k1.channel = c2

提示：输出的本地目录必须是已经存在的目录，如果该目录不存在，并不会创建新的目录。

（5）执行配置文件

分别启动对应的 flume 进程：flume-flume-dir，flume-flume-hdfs，flume-file-flume。

[atguigu@hadoop102 flume]$ bin/flume-ng agent --conf conf/ --name 
a3 --conf-file job/group1/flume-flume-dir.conf
[atguigu@hadoop102 flume]$ bin/flume-ng agent --conf conf/ --name 
a2 --conf-file job/group1/flume-flume-hdfs.conf
[atguigu@hadoop102 flume]$ bin/flume-ng agent --conf conf/ --name 
a1 --conf-file job/group1/flume-file-flume.conf

（6）启动 Hadoop 和 Hive

[atguigu@hadoop102 hadoop-2.7.2]$ sbin/start-dfs.sh
[atguigu@hadoop103 hadoop-2.7.2]$ sbin/start-yarn.sh
[atguigu@hadoop102 hive]$ bin/hive
hive (default)>

（7）检查 HDFS 上数据

（8）检查/opt/module/datas/flume3 目录中数据

[atguigu@hadoop102 flume3]$ ll
总用量 8
-rw-rw-r--. 1 atguigu atguigu 5942 5 月 22 00:09 1526918887550-3