当前位置: 首页 > article >正文

Flume source类型和使用案例(exec,spoodir,Taildir)

目录

  • exec
    • 定义
    • 功能
    • 配置文件
      • 配置参数
      • exec-logger.conf
    • 执行脚本
    • 效果
  • spooldir
    • 定义
    • 功能
    • 配置文件
      • 配置参数
      • spooldir-logger.conf
    • 执行脚本
    • 效果
    • 说明
  • Taildir
    • 定义
    • 功能
    • 配置文件
      • 配置参数
      • taildir-logger.conf
    • 执行脚本
    • 说明


exec

定义

exec是Flume中的一种Source组件,用于执行用户配置的Unix命令,并从命令的标准输出中采集数据。

功能

exec Source能够持续监控命令的输出,并将输出数据转换为Flume的Event格式,然后通过Channel传递给Sink进行处理。

配置文件

配置参数

在这里插入图片描述

exec-logger.conf

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

# Describe/configure the source
a1.sources.r1.type = exec
# 持续追踪/root/log.txt文件的新增内容
a1.sources.r1.command = tail -F /root/log.txt
# 使用Bash shell执行命令
a1.sources.r1.shell = /bin/bash -c
# 每次批量处理3个事件
a1.sources.r1.batchsize=3
# 如果5秒内没有足够的事件达到batchsize,则触发批次处理(等待时长)
a1.sources.r1.batchTimeout = 5000

# Describe the sink
a1.sinks.k1.type = logger

# 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


执行脚本

flume-ng agent --conf ./ --name a1 --conf-file exec-logger.conf

–conf 指定了 Flume 配置文件的目录。
–name 指定了 agent 的名称,这里为 a1。
–conf-file 指定了 agent 的具体配置文件,这里为 exec-logger.conf。

效果

红框为原有的内容
使用vi语句新增内容“wunaiieq”后输出如下(全部显示)
在这里插入图片描述
使用echo在文件后增加内容“new line”,结果仅增加新的一行
在这里插入图片描述

spooldir

定义

监控指定目录(spool directory)下新产生的文件,并读取这些文件的内容。

功能

持续监控指定目录(spool directory)下新产生的文件。
能够递归地检查目录及其子目录中的所有文件(具体取决于 Flume 的版本和配置)。
当检测到新文件时,自动读取文件内容。

配置文件

配置参数

在这里插入图片描述

spooldir-logger.conf

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

# Describe/configure the source
a1.sources.r1.type = spooldir
# 监控目录(spooldir)
a1.sources.r1.spoolDir = /opt/module/jar/test
# 文件后缀(fileSuffix)
a1.sources.r1.fileSuffix= .wunaiieq
# 忽略模式(ignorePattern)
a1.sources.r1.ignorePattern=([^ ]*\.tmp)
# 文件头(fileHeader):true,表示在事件头中添加文件的绝对路径。
a1.sources.r1.fileHeader = true

# Describe the sink
a1.sinks.k1.type = logger

# 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

执行脚本

flume-ng agent --conf ./ --name a1 --conf-file spooldir-logger.conf

效果

创建文件20240216.txt,写入aa
结果如下
xftp文件显示
在这里插入图片描述
输出日志
在这里插入图片描述

说明

spooldir主要用途是处理文本文件或可以逐行读取的文件类型。其他文件如.jar什么的不可以,产生如下的报错

2024-12-06 09:33:03,486 ERROR source.SpoolDirectorySource: FATAL: Spool Directory source r1: { spoolDir: /opt/module/jar/test }: Uncaught exception in SpoolDirectorySource thread. Restart or reconfigure Flume to continue processing.
java.nio.charset.MalformedInputException: Input length = 1
at java.nio.charset.CoderResult.throwException(CoderResult.java:281)
at org.apache.flume.serialization.ResettableFileInputStream.readChar(ResettableFileInputStream.java:283)
at org.apache.flume.serialization.LineDeserializer.readLine(LineDeserializer.java:132)
at org.apache.flume.serialization.LineDeserializer.readEvent(LineDeserializer.java:70)
at org.apache.flume.serialization.LineDeserializer.readEvents(LineDeserializer.java:89)
at org.apache.flume.client.avro.ReliableSpoolingFileEventReader.readDeserializerEvents(ReliableSpoolingFileEventReader.java:403)
at org.apache.flume.client.avro.ReliableSpoolingFileEventReader.readEvents(ReliableSpoolingFileEventReader.java:378)
at org.apache.flume.source.SpoolDirectorySource S p o o l D i r e c t o r y R u n n a b l e . r u n ( S p o o l D i r e c t o r y S o u r c e . j a v a : 263 ) a t j a v a . u t i l . c o n c u r r e n t . E x e c u t o r s SpoolDirectoryRunnable.run(SpoolDirectorySource.java:263) at java.util.concurrent.Executors SpoolDirectoryRunnable.run(SpoolDirectorySource.java:263)atjava.util.concurrent.ExecutorsRunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access 301 ( S c h e d u l e d T h r e a d P o o l E x e c u t o r . j a v a : 180 ) a t j a v a . u t i l . c o n c u r r e n t . S c h e d u l e d T h r e a d P o o l E x e c u t o r 301(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor 301(ScheduledThreadPoolExecutor.java:180)atjava.util.concurrent.ScheduledThreadPoolExecutorScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)

Taildir

定义

用于实时监控目录下的文件变化,并将这些文件的内容作为数据源发送到Flume的Channel中供后续处理或转发。

功能

能够实时监控指定目录下的文件变化,包括新文件的创建、现有文件的修改等。
它支持多种格式的日志文件,如*.log,并且兼容多种编码方式,非常适合用于系统和应用日志的实时采集和分析。
Taildir Source支持断点续传功能,即使Flume重启,它也能从上次读取的位置继续采集数据,确保数据不丢失。
Taildir Source支持多线程异步处理,能够显著提高并发性能,确保在高负载情况下也能稳定工作。

配置文件

配置参数

在这里插入图片描述

taildir-logger.conf

# a1--agent
# all source of a1
a1.sources = r1
# all sinks of a1
a1.sinks = k1
# all channels of a1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = taildir
#  指定源要监控的文件组为fg1和fg2。
a1.sources.r1.filegroups = fg1 fg2
# 定义文件组fg1包含/opt/module/jar/目录下所有以.txt结尾的文件。
a1.sources.r1.filegroups.fg1 = /opt/module/jar/.*\.txt
# 定义文件组fg2包含/opt/module/jar/目录下所有以.conf结尾的文件。
a1.sources.r1.filegroups.fg2 = /opt/module/jar/.*\.log
# 指定记录文件读取位置的文件路径,用于在agent重启时能够从上次读取的位置继续读取。
a1.sources.r1.positionFile = /opt/module/flume/jobs/taildir/taildir.json

# Describe the sink
a1.sinks.k1.type = logger

# 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

执行脚本

flume-ng agent --conf ./ --name a1 --conf-file taildir-logger.conf

说明

a1.sources.r1.positionFile = /opt/module/flume/jobs/taildir/taildir.json

这里的json文件可以说是运行时的日志文件,断点后读取文件,记录变动情况


http://www.kler.cn/a/427620.html

相关文章:

  • 【数字电路与逻辑设计】实验五 4人表决器
  • C#设计模式--适配器模式(Adapter Pattern)
  • WordPress XStore Elementor 前端与编辑器内容不同步的问题
  • 数据科学和机器学习中的多重共线性:隐藏的威胁及其解决方法
  • ASP.NET Core Web 项目的部署:选择 IIS 还是 Kestrel?
  • 【周易哲学】生辰八字入门讲解(七)
  • 【工具】音频文件格式转换工具
  • mongoDB 8.0 群集搭建和遇到的问题
  • OpenCV 图像基本操作
  • Oracle数据库使用dblink是时出现 ORA-12170:TNS:连接超时
  • 【Springboot3+vue3】从零到一搭建Springboot3+vue3前后端分离项目之后端环境搭建
  • 基于单片机的自限位电机正反转控制电路
  • 基于合成错误增强的标签精细化网络用于医学图像分割|文献速递-生成式模型与transformer在医学影像中的应用
  • 【Elasticsearch】Docker安装和基本概念
  • DevOps工程师成长路线图
  • 面试小札:线程池
  • 下载谷歌浏览器的官方离线安装包
  • 【论文解读】MD-VQA: Multi-Dimensional Quality Assessment for UGC Live Videos
  • TCP编程案例
  • 基于Matlab的卷积神经网络(CNN)苹果分级检测系统