中国移动 acm facebook go cmake scope requirejs h5表格模板 网校直播系统 linux超级用户 python与机器学习 edate函数的使用方法 当前时间减一天 新手学c还是java plsql连接mysql 汇编语言clr python开发教程 python正则表达 python查找指定字符 java语言基础教程 javaswitch语句 java中的string java系统时间 java的输入 java架构 java日期格式化 java读取文本 java格式化日期 街头篮球辅助 sql行转列 tftpd64 rar去广告 电脑基础 vs2003 steam怎么卸载 tomcat修改端口 php取整 工程地质手册 renderpartial dnf胜利的证明有什么用
当前位置: 首页 > 学习教程  > 编程语言

flume监听文件追加内容并上传到HDFS

2020/10/8 18:38:59 文章标签:

flume监听文件追加内容并上传到HDFS一、监听单个追加文件1.1 需求1.2 准备工作1.3 编写配置文件1.4 运行flume二 监听目录多个追加文件2.1 需求2.2 编写配置文件2.3 启动flume一、监听单个追加文件 1.1 需求 监控test.log日志文件,将日志文件追加的内容上传到hdfs…

flume监听文件追加内容并上传到HDFS

  • 一、监听单个追加文件
    • 1.1 需求
    • 1.2 准备工作
    • 1.3 编写配置文件
    • 1.4 运行flume
  • 二 监听目录多个追加文件
    • 2.1 需求
    • 2.2 编写配置文件
    • 2.3 启动flume

一、监听单个追加文件

1.1 需求

监控test.log日志文件,将日志文件追加的内容上传到hdfs

1.2 准备工作

Flume要想将数据输出到HDFS,必须持有Hadoop相关jar包,将对应jar复制到 /flume/lib 文件夹下
这里使用hadoop版本:hadoop-2.7.6

##需要的Jar
hadoop-common-2.7.6.jar
commons-configuration-1.6.jar
hadoop-auth-2.7.6.jar
hadoop-hdfs-2.7.6.jar
commons-io-2.4.jar
htrace-core-3.1.0-incubating.jar

##对应hadoop目录
/opt/apps/hadoop-2.7.6/share/hadoop/common
/opt/apps/hadoop-2.7.6/share/hadoop/common/lib
/opt/apps/hadoop-2.7.6/share/hadoop/common/lib
/opt/apps/hadoop-2.7.6/share/hadoop/hdfs
/opt/apps/hadoop-2.7.6/share/hadoop/hdfs/lib
/opt/apps/hadoop-2.7.6/share/hadoop/hdfs/lib

1.3 编写配置文件

在flume路径下的job路径中新建flume-hdfs.conf

vim /opt/apps/apache-flume-1.9.0-bin/job/flume-hdfs.conf

此文件编写内容如下

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

# Describe/configure the sources
# exec即execute执行命令
a1.sources.r1.type = exec
# 要执行的命令
a1.sources.r1.command = tail -F /data/test.log
# 执行shell脚本的绝对路径
a1.sources.r1.shell = /bin/bash -c

# Describe the sink
a1.sinks.k1.type = hdfs
# 上传到hdfs的路径
a1.sinks.k1.hdfs.path = hdfs://slave1:9000/flume/%Y%m%d%H
# 文件前缀
a1.sinks.k1.hdfs.filePrefix = logs-
# 是否按时间新建文件夹
a1.sinks.k1.hdfs.round = true
# 定义多久新建文件夹
a1.sinks.k1.hdfs.roundValue = 1
# 重新定义时间单位
a1.sinks.k1.hdfs.roundUnit = hour
# 是否使用本地时间戳
a1.sinks.k1.hdfs.useLocalTimeStamp = true
# flush到hdfs需要积攒event的数量
a1.sinks.k1.hdfs.batchSize = 50
# 设置文件类型
a1.sinks.k1.hdfs.fileType = DataStream
# 多久生成一个新的文件(单位:秒)
a1.sinks.k1.hdfs.rollInterval = 60
# 接收多少数据生成一个新文件(单位:字节)
a1.sinks.k1.hdfs.rollSize = 134217700
# 文件的滚动与event数量无关
a1.sinks.k1.hdfs.rollCount = 0
# 最小冗余数
a1.sinks.k1.hdfs.minBlockReplicas = 1

# 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

注意:
1.a1.sinks.k1.hdfs.path需要和hadoop的core-site.xml指定hdfs路径一样
2.tail -f 和tail -F 区别
tail -f
等同于–follow=descriptor,根据文件描述符进行追踪,当文件改名或被删除,追踪停止
tail -F
等同于–follow=name --retry,根据文件名进行追踪,并保持重试,即该文件被删除或改名后,如果再次创建相同的文件名,会继续追踪

1.4 运行flume

因为需要落地到hdfs,要先确保hadoop集群

在这里插入图片描述

在flume路径下

bin/flume-ng agent --conf conf/ --name a1 --conf-file job/flume-hdfs.conf   -Dflume.root.logger=INFO,console 
#写入数据到test.log
echo "123" >> test.log

然后查看hdfs目录
在这里插入图片描述

二 监听目录多个追加文件

2.1 需求

监听/data/flume目录下多个追加文件并上传到HDFS,同时实现断点续传功能。

Exec source适用于监控一个实时追加的文件,不能实现断电续传;Spooldir Source适合用于同步新文件,但不适合对实时追加日志的文件进行监听并同步;而Taildir Source适合用于监听多个实时追加的文件,并且能够实现断点续传。

2.2 编写配置文件

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 = /data/flume/tail_dir.json
a3.sources.r3.filegroups = f1 f2
a3.sources.r3.filegroups.f1 = /data/flume/.*file.*
a3.sources.r3.filegroups.f2 = /data/flume/.*log.*

# Describe the sink
a3.sinks.k3.type = hdfs
a3.sinks.k3.hdfs.path = hdfs://slave1:9000/flume/%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 = 50
#设置文件类型,可支持压缩
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

2.3 启动flume

bin/flume-ng agent --conf conf/ --name a3 --conf-file job/flume-taildir-hdfs.conf   -Dflume.root.logger=INFO,console 

在监控目录下创建文件并追加

touch file1.txt
touch file2.txt
touch file3.txt
echo 123 >>file1.txt
echo 4545 >>file2.txt

查看效果
在这里插入图片描述

##查看tail_dir.json
cat tail_dir.json
[{"inode":1452290,"pos":4,"file":"/data/flume/file1.txt"},{"inode":1452302,"pos":5,"file":"/data/flume/file2.txt"},{"inode":1452322,"pos":0,"file":"/data/flume/file3.txt"}]

本文链接: http://www.dtmao.cc/news_show_250084.shtml

附件下载

相关教程

    暂无相关的数据...

共有条评论 网友评论

验证码: 看不清楚?