Flume基础

安装

1. 下载apache-flume-1.11.0-bin.tar.gz。

2. 解压：

tar -zxvf apache-flume-1.11.0-bin.tar.gz

1. 重新命名下

mv apache-flume-1.11.0-bin apache-flume-1.11.0

1. 开启环境配置。将 flume/conf 目录下的 flume-env.sh.template 文件修改为 flume-env.sh。

mv flume-env.sh.template flume-env.sh

1. 配置 flume-env.sh 文件，将 LInux 系统的 jdk 的路径写到其中。

export JAVA_HOME=/usr/local/java/jdk1.8.0_151

案例

说明

Exec Source 适用于监控一个实时追加的文件，但不能保证数据不丢失；Spooldir Source 能够保证数据不丢失，且能实现断点续传，但延迟较高，不能实时监控；而 Taildir Source 既能实现断点续传，又可以保证数据不丢失，还能够进行实时监控。

监控端口数据

使用 Flume 监听一个端口，收集该端口数据，并打印到控制台。

1. 在apache-flume-1.11.0目录下新增job目录，并创建agent配置文件

mkdir job
cd job
vim flume-netport-logger.conf

1. flume-netport-logger.conf配置文件增加内容。

# example.conf: A single-node Flume configuration
# Name the components on this agent
# 为组件命名,agent的名称是a1
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
# 配置sources
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
# 监听44444端口
a1.sources.r1.port = 44444

# Describe the sink
# 配置sinks
# 输出到logger
a1.sinks.k1.type = logger

# Use a channel which buffers events in memory
# 配置channels
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
# 将source和sink绑定到channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

1. 启动

bin/flume-ng agent -c conf -f job/flume-netport-logger.conf -n a1 -Dflume.root.logger=INFO,console

1. telnet 发送测试

1. 结果验证

查看flume.log

监控单个追加文件

1. job目录下创建flume-file-logger.conf文件

vim flume-file-logger.conf

1. 添加内容

# Name the components on this agent
a2.sources = r2
a2.sinks = k2
a2.channels = c2

# Describe/configure the source
a2.sources.r2.type = exec
a2.sources.r2.command = tail -F /data/test/test.log


# Describe the sink
a2.sinks.k2.type = logger

# Use a channel which buffers events in memory
a2.channels.c2.type = memory
a2.channels.c2.capacity = 1000
a2.channels.c2.transactionCapacity = 100

# Bind the source and sink to the channel
a2.sources.r2.channels = c2
a2.sinks.k2.channel = c2

1. 运行flume

bin/flume-ng agent -c conf -f job/flume-file-logger.conf -n a2 -Dflume.root.logger=INFO,console

1. 往test.log文件写入内容

2. 查看flume日志

监控目录下多个新文件,输出到hdfs

1. 创建agent配置文件

vim flume-dir-hdfs.conf

1. 配置文件内容

# Name the components on this agent
a3.sources = r3
a3.sinks = k3
a3.channels = c3

# Describe/configure the source
a3.sources.r3.type = spooldir
a3.sources.r3.spoolDir = /opt/module/flume/upload
a3.sources.r3.fileSuffix = .COMPLETED
a3.sources.r3.fileHeader = true
#忽略所有以.tmp 结尾的文件，不上传
a3.sources.r3.ignorePattern = ([^ ]*\.tmp)

# Describe the sink
a3.sinks.k3.type = hdfs
a3.sinks.k3.hdfs.path =  hdfs://master: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 = 10
#设置文件类型，可支持压缩
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

1. 启动监控文件夹命令

bin/flume-ng agent -c conf -f job/flume-dir-hdfs.conf -n a3

1. 在/opt/module/flume/upload新增文件

2. 查看hdfs目录是否有文件生成

需要注意的是：linux和权限和hdfs的权限，当以root用户开启flume监控的时候，由于hdfs写入需要时hdfs用户，因此此时flume会提示权限不足。

解决方法：用hdfs的用户去开启flume监控。

监控目录下的多个追加文件

1. 创建配置文件 flume-taildir-hdfs.conf

vim flume-taildir-hdfs.conf

1. 添加内容

# Name the components on this agent
a4.sources = r4
a4.sinks = k4
a4.channels = c4

# Describe/configure the source
a4.sources.r4.type = TAILDIR
a4.sources.r4.positionFile = /opt/module/flume/postion/position.json
a4.sources.r4.filegroups = f1 f2
a4.sources.r4.filegroups.f1 = /opt/module/flume/files/file1.txt
a4.sources.r4.filegroups.f2 = /opt/module/flume/files/file2.txt

# Describe the sink
a4.sinks.k4.type = hdfs
a4.sinks.k4.hdfs.path = hdfs://master:9000/flume/%Y%m%d/%H
#上传文件的前缀
a4.sinks.k4.hdfs.filePrefix = upload- 
#是否按照时间滚动文件夹
a4.sinks.k4.hdfs.round = true
#多少时间单位创建一个新的文件夹
a4.sinks.k4.hdfs.roundValue = 1
#重新定义时间单位
a4.sinks.k4.hdfs.roundUnit = hour
#是否使用本地时间戳
a4.sinks.k4.hdfs.useLocalTimeStamp = true
#积攒多少个 Event 才 flush 到 HDFS 一次
a4.sinks.k4.hdfs.batchSize = 10
#设置文件类型，可支持压缩
a4.sinks.k4.hdfs.fileType = DataStream
#多久生成一个新的文件
a4.sinks.k4.hdfs.rollInterval = 60
#设置每个文件的滚动大小大概是 128M
a4.sinks.k4.hdfs.rollSize = 134217700
#文件的滚动与 Event 数量无关
a4.sinks.k4.hdfs.rollCount = 0

# Use a channel which buffers events in memory
a4.channels.c4.type = memory
a4.channels.c4.capacity = 1000
a4.channels.c4.transactionCapacity = 100

# Bind the source and sink to the channel
a4.sources.r4.channels = c4
a4.sinks.k4.channel = c4

1. 启动监控

bin/flume-ng agent -c conf/ -f job/flume-taildir-hdfs.conf -n a4

1. 追加内容。/opt/module/flume/files

echo hello >> file1.txt
echo hello >> file2.txt

1. 验证文件的生成