Flume采集文件内容上传至HDFS

落日映苍穹つ 2022-08-30 12:56 144阅读 0赞

需求:采集目录中已有的文件内容，存储到HDFS 分析:source是要基于目录的，channel建议使用file，可以保证不丢数据，sink使用hdfs  
下面要做的就是配置Agent了，可以把example.conf拿过来修改一下，新的文件名为`file-to-hdfs.conf`

# Name the components on this agent
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1

先配置一下source

# Describe/configure the source
    a1.sources.r1.type = spooldir
    a1.sources.r1.spoolDir = /data/log/studentDir

接下来是channel了  
channel在这里使用基于文件的，可以保证数据的安全性 如果针对采集的数据，丢个一两条对整体结果影响不大，只要求采集效率，那么这个时候完全可以使用基于内存的channel  
主要配置checkpointDir和dataDir，因为这两个目录默认会在用户家目录下生成， 建议修改到其他地方  
checkpointDir是存放检查点目录  
data是存放数据的目录

# Use a channel which buffers events in memory
    a1.channels.c1.type = file
    a1.channels.c1.checkpointDir = /data/soft/apache-flume-1.9.0-bin/data/studentDir/checkpoint
    a1.channels.c1.dataDirs = /data/soft/apache-flume-1.9.0-bin/data/studentDir/data

最后是sink 因为要向hdfs中输出数据，所以可以使用hdfssink

一般在这我们需要设置`writeFormat`和`fileType`这两个参数  
默认情况下writeFormat的值是Writable，建议改为Text，看后面的解释，如果后期想使用hive或者 impala操作这份数据的话，必须在生成数据之前设置为Text，Text表示是普通文本数据  
fileType默认是SequenceFile，还支持DataStream 和 CompressedStream ，DataStream 不会对输出 数据进行压缩，CompressedStream 会对输出数据进行压缩，在这里我们先不使用压缩格式的，所以选 择DataStream

除了这些参数以外，还有三个也比较重要 hdfs.rollInterval、hdfs.rollSize和hdfs.rollCount

*  hdfs.rollInterval默认值是30，单位是秒，表示hdfs多长时间切分一个文件，因为这个采集程序是一 直运行的，只要有新数据，就会被采集到hdfs上面，hdfs默认30秒钟切分出来一个文件，如果设置 为0表示不按时间切文件
 *  hdfs.rollSize默认是1024，单位是字节，最终hdfs上切出来的文件大小都是1024字节，如果设置为0 表示不按大小切文件
 *  hdfs.rollCount默认设置为10，表示每隔10条数据切出来一个文件，如果设置为0表示不按数据条数切文件

这三个参数，如果都设置的有值，哪个条件先满足就按照哪个条件都会执行。  
在实际工作中一般会根据时间或者文件大小来切分文件，我们之前在工作中是设置的时间和文件大小相结合，时间设置的是一小时，文件大小设置的128M，这两个哪个满足执行哪个

所以针对hdfssink的配置最终是这样的

a1.sinks.k1.type = hdfs
    a1.sinks.k1.hdfs.path = hdfs://192.168.182.100:9000/flume/studentDir
    a1.sinks.k1.hdfs.filePrefix = stu-
    a1.sinks.k1.hdfs.fileType = DataStream
    a1.sinks.k1.hdfs.writeFormat = Text
    a1.sinks.k1.hdfs.rollInterval = 3600
    a1.sinks.k1.hdfs.rollSize = 134217728
    a1.sinks.k1.hdfs.rollCount = 0

最后把组件连接到一起

# Bind the source and sink to the channel
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

下面就可以启动agent了，在启动agent之前，先初始化一下测试数据 创建/data/log/studentDir目录，然后在里面添加一个文件，class1.dat class1.dat中存储的是学生信息，学生姓名、年龄、性别

[root@bigdata02 ~]# mkdir -p /data/log/studentDir
    [root@bigdata02 ~]# cd /data/log/studentDir
    [root@bigdata02 studentDir]# more class1.dat
    jack    18      male
    jessic  20      female
    tom     17      male

启动Hadoop集群

[root@bigdata01 ~]# cd /data/soft/hadoop-3.2.0

启动Agent，使用在前台启动的方式，方便观察现象  
但是发现在启动的时候报错，提示找不到SequenceFile，但是我们已经把fileType改为了DataStream， 但是Flume默认还是会加载这个类  
`Caused by: java.lang.ClassNotFoundException: org.apache.hadoop.io.SequenceFile$CompressionT ype`

就算你把SequenceFile相关的jar包都拷贝到flume的lib目录下解决了这个问题，但是还是会遇到找不到 找不到HDFS这种文件类型，还是缺少hdfs相关的jar包  
`No FileSystem for scheme: hdfs`

当然这个问题也可以通过拷贝jar包来解决这个问题，但是这样其实太费劲了，并且后期我们有很大可能 需要在这个节点上操作HDFS，所以其实最简单直接的方法就是把这个节点设置为hadoop集群的一个客 户端节点，这样操作hdfs就没有任何问题了。

我们直接把集群中修改好配置的 hadoop目录远程拷贝到bigdata02上就可以了

> 注意:还需要修改环境变量，配置HADOOP\_HOME，否则启动Agent的时候还是会提示找不到 SequenceFile

修改完后Agent就可以正常启动了

此时发现文件已经生成了，只不过默认情况下现在的文件是`.tmp`结尾的，表示它在被使用，因为Flume只要采集到数据就会向里面写，这个后缀默认是由`hdfs.inUseSuffix`参数来控制的

文件名上还拼接了一个当前时间戳，这个是默认文件名的格式，当达到文件切割时机的时候会给文件改名字，去掉.tmp  
这个文件现在也是可以查看的，里面的内容其实就是class1.dat文件中的内容

[root@bigdata04 ~]# hdfs dfs -cat hdfs://192.168.182.100:9000/flume/studentDi
    jack    18      male
    jessic  20      female
    tom     17      male

所以此时Flume就会监控linux中的/data/log/studentDir目录，当发现里面有新文件的时候就会把数据采集过来。