再战flume关于实时监控多个目录下多个追加文件原理解释配置文件解读开始前可能遇到的bug

62 阅读 0 评论 41 点赞

我是靠谱客的博主义气翅膀，最近开发中收集的这篇文章主要介绍再战flume关于实时监控多个目录下多个追加文件原理解释配置文件解读开始前可能遇到的bug，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

我们上次练习的source用的是简单的netcat，除了它之外比较常见的还用：
Exec source：适用于监控一个实时追加的文件，不能实现断点续传；
Spooldir Source：适合用于同步新文件，但不适合对实时追加日志的文件进行监听并同步；
以上两种source都有各种缺陷，所以这次我们将使用更强的大Taildir Source来实验
Taildir Source适合用于监听多个实时追加的文件，并且能够实现断点续传。

原理解释

首先介绍下Taildir Source断点续传的原理:
我们用flume划定了一个目录后，这个目录每当有新数据进入，我们就通过flume这个进程来不断的写入到hdfs中；假设有一批数据，有百分之十已经写入到flume划定的目录中时，此时flume正在好好做着工作的进程挂掉了，但是这批数据还在不断写入；当我们再次开启flume时，数据已经写到百分之五十了。而Taildir Source由于有断点续传的功能，它不会直接从百分之五十的位置继续读（这样会丢数据），而是从百分之十的位置继续读。原因是Taildir Source配置了一个地址用来存它指引，而当数据写到百分之十的时候，flume挂掉之前，Taildir Source会有一个指引标志了数据flume读到哪了，当flume重启后会继续按着索引位置继续读数据。

然后还要说一下它在往hdfs上写的方法，因为有时数据高峰来的数据会很多，但是有时候数据可能长时间内来的很少，我们不能一直保持将数据存在一个文件内，所以要不断的滚动生成文件来存储在hdfs中。
而hdfs sink准备了三个参数，也就是滚动的临界条件，底下我们在详细介绍，当数据来了之后，我们先写在一个以.tmp结尾的文件中，当达到了临界条件，会生成一个数据文件写入到hdfs中，然后再生成一个.tmp结尾的文件，继续写入数据。

配置文件解读

明白了原理我们就开始创建这次flume的配置文件

同样我们先去找Taildir Source的官网介绍
在这里插入图片描述
可以看见，这里可以放一个file集合，也就是我们实现监控多个目录下文件的原因
所以配置文件应该如下方法写
并且我们可以写一个positionFile来指定我们记录读取数据到哪的指引所存的地址

如上，我们将帮助断点续传的指引存到tail_dir.json
然后我们来看一下sink怎么写的，首先看下官网的介绍
在这里插入图片描述
hdfs.path不用所说，就是存储在hdfs上的目录
然后还需要介绍三个很重要的属性如下

rollInterval代表了多长时间我们会滚动生成一个新的文件，比如我们先生成一个.tmp文件然后不断往里写数据，当没达到其他临界条件后，但到了rollInterval规定的临界时间后，结束.tmp，滚动生成新的文件
hdfs.rollSize 代表的临界条件为，当.tmp文件多大后开始滚动
rollCount代表接受多少事务后滚动，一般设置为0，也就是不适用这个属性（0 = disable automatic closing of idle files)）

然后可以写出配置文件如下

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:8020/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 = 30
#设置每个文件的滚动大小大概是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

开始前可能遇到的bug

我们的sink储存在hdfs的路径即使不存在，在我们开始flume并且传数据后，会自动再hdfs中创建路径的目录
但是注意！，source需要的路径必须在开始flume之前就创建好，否则flume会报错

所以第一步，我们需要先创建传数据给source的目录和放json文件的目录

在这里插入图片描述
然后写配置文件放到job中，然后开启flume
bin/flume-ng agent --conf conf/ --name a3 --conf-file job/flume-taildir-hdfs.conf
同时再开一个窗口进入到files目录下，往里传数据echo hello >> file1.txt
此时可以再hdfs中看见出现了存储目录，且格式为配置文件中所写
在这里插入图片描述
此时我们换一个文件夹再加一次数据，并且操作快点，可以看见tmp
echo atguigu >> file2log.txt

然后30s后，达到临界条件，这个tmp结尾的文件就会变成正式文件