Flume使用案例之实时读取目录到HDFS

楼高2年前技术文章556

目标:使用flume监听整个目录的文件

分步实现

1. 创建配置文件flume-dir.conf

#1 Agent

a3.sources = r3

a3.sinks = k3

a3.channels = c3

 

#2 source

a3.sources.r3.type = spooldir

a3.sources.r3.spoolDir = /opt/module/flume1.8.0/upload

a3.sources.r3.fileSuffix = .COMPLETED

a3.sources.r3.fileHeader = true

#忽略所有以.tmp结尾的文件,不上传

a3.sources.r3.ignorePattern = ([^ ]*\.tmp)

 

# 3 sink

a3.sinks.k3.type = hdfs

a3.sinks.k3.hdfs.path = hdfs://dtstack_hdfs:9000/flume/%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

#积攒多少个EventflushHDFS一次

a3.sinks.k3.hdfs.batchSize = 100

#设置文件类型,可支持压缩

a3.sinks.k3.hdfs.fileType = DataStream

#多久生成一个新的文件

a3.sinks.k3.hdfs.rollInterval = 600

#设置每个文件的滚动大小大概是128M

a3.sinks.k3.hdfs.rollSize = 134217700

#文件的滚动与Event数量无关

a3.sinks.k3.hdfs.rollCount = 0

#最小副本数

a3.sinks.k3.hdfs.minBlockReplicas = 1

 

# 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. 执行测试:执行如下脚本后,请向upload文件夹中添加文件试试

/opt/module/flume1.8.0/bin/flume-ng agent \

--conf /opt/module/flume1.8.0/conf/ \

--name a3 \

--conf-file /opt/module/flume1.8.0/jobconf/flume-dir.conf

提示: 在使用Spooling Directory Source

1) 不要在监控目录中创建并持续修改文件

2) 上传完成的文件会以.COMPLETED结尾

3) 被监控文件夹每500毫秒扫描一次文件变动

相关文章

Hive 重新编译-解决Tez JobName的问题

Hive 重新编译-解决Tez JobName的问题

本文采用linux编译首先下载源码https://dlcdn.apache.org/hive/hive-3.1.2/apache-hive-3.1.2-src.tar.gz源码位置ql/src/jav...

CDH实操--集群卸载

CDH实操--集群卸载

1、停止正在运行的服务a、控制台停止集群服务b、控制台停止Cloudera Management Servicec、命令行停止cm服务systemctl stop cloudera-scm-agent...

Ranger-hdfs插件部署

Ranger-hdfs插件部署

部署在两个namenode节点解压插件cd /opt/hadooptar -xzvf ranger-2.4.0-hdfs-plugin.tar.gz -C /opt/cd /opt/ranger-2....

PostgreSQL 基于时间点恢复

前言本篇文章介绍 PostgreSQL 基于时间点恢复(point-in-time-recover)需要的条件及恢复过程,属于操作说明。1. WAL 日志WAL(Write Ahead Log) 日志...

SpringBootWeb 篇-深入了解 SpringBoot + Vue 的前后端分离项目部署上线与 Nginx 配置文件结构(3)

SpringBootWeb 篇-深入了解 SpringBoot + Vue 的前后端分离项目部署上线与 Nginx 配置文件结构(3)

4.0 SpringBoot 后端代码进行打包上传服务器        将 SpringBoot 项目进行打包:      &nb...

压测实操--produce压测方案

压测实操--produce压测方案

环境信息:操作系统centos7.9,kafka版本为hdp集群中的2.0版本。 Producer相关参数使用Kafka自带的kafka-producer-perf-test.sh脚本进行压测,该脚本...

发表评论    

◎欢迎参与讨论,请在这里发表您的看法、交流您的观点。