flume Source志SpoolDir

Posted 2023-01-06 宝哥大数据[离职找工作中，大佬帮内推下]

二、flume监控某个目录，将数据发送kafka

2.1、flume source 使用SpoolDir 监控一个目录下的文件内容

SpoolDir监控目录下文件， 处理后的文件修改文件名

问题1： 如果上传来的文件,还没有上传完，spoolDir扫描到改文件，将会报错

所以此处我将监控前一个小时目录，

#!/bin/bash
#date -d "-10 min"   +%H
pred=`date -d "-2 hour" +%Y%m%d`
d=`date -d "-1 hour" +%Y%m%d`
preh=`date -d "-2 hour" +%H`
h=`date -d "-1 hour" +%H`
echo $d-$h
cd /opt/bigdata/apache-flume-1.5.2-bin/conf/
sed -i "s/idc_bakupload\\/$pred\\/$preh\\/idc/idc_bakupload\\/$d\\/$h\\/idc/g" dir-flume-conf.properties

问题2、spoolDir 无法监控多级目录

启动flume

./bin/flume-ng agent --conf conf/ --conf-file conf/file-flume-conf.properties --name a1 

完整的配置

# 监听ftp日志
#agent的名字a1
a1.sources = src
a1.channels = chl
a1.sinks = sk


#定义source， source使用spooldir， 监控ftp日志
a1.sources.src.type=spooldir
#监控目录
a1.sources.src.spoolDir=/hadoop/ftp/idc_bakupload/20180307/22/idc/
#忽略的文件
a1.sources.src.ignorePattern = ^(.)*\\\\.AVL\\\\.(.)*$
#处理后的文件，添加后缀
a1.sources.src.fileSuffix = .bak


#定义channel， 使用memory 作为channel
a1.channels.chl.type = memory
a1.channels.chl.capacity = 100000
a1.channels.chl.transactionCapacity = 10000

#定义sink， 输出到控制台
a1.sinks.sk.type = com.chb.test.sink.MyKafkaSink

#定义sink， source 与channel的关系
#注意sink后面是channel, 而不是s
a1.sinks.sk.channel = chl
a1.sources.src.channels = chl