Flume详解——介绍、部署与使用

news2025/3/19 1:35:15

1. Flume 简介

Apache Flume 是一个专门用于高效地 收集、聚合、传输 大量日志数据的 分布式、可靠 的系统。它特别擅长将数据从各种数据源（如日志文件、消息队列等）传输到 HDFS、HBase、Kafka 等大数据存储系统。

特点：

  • 可扩展：支持大规模数据传输，灵活扩展
  • 容错性：支持数据恢复和失败重试，确保数据不丢失
  • 多种数据源：支持日志文件、网络数据、HTTP请求、消息队列等多种来源
  • 流式处理：数据边收集边传输，支持实时传输

2. Flume 架构

Flume 的核心架构由三大组件构成，理解它们对掌握 Flume 的原理至关重要：

2.1 Source（来源）

负责从数据源获取数据，比如：

  • taildir（监听日志文件）
  • exec（执行命令读取数据）
  • kafka（从 Kafka 消费数据）
  • netcat（监听端口接收数据）

2.2 Channel（通道）

作为 缓冲区，临时存储数据，支持两种常见类型：

  • Memory Channel（内存通道）：速度快，但重启可能丢数据
  • File Channel（文件通道）：写入磁盘，保证数据持久性

2.3 Sink（下游输出）

负责将数据写入目标位置，支持：

  • HDFS（写入 Hadoop 分布式文件系统）
  • HBase（写入 HBase 数据库）
  • Kafka（推送到 Kafka）
  • ElasticSearch（支持实时检索）

补充组件：

  • Sink Processor：管理多个 Sink，支持负载均衡、故障转移
  • Interceptor：在数据进入 Channel 前拦截处理，比如格式转换、过滤数据等

3. Flume 数据流动原理

数据在 Flume 中是按事件 (Event) 传输的，基本流程如下：

1️⃣ Source 从外部采集数据，将每条数据封装为一个 Event
2️⃣ Event 进入 Channel 暂存
3️⃣ Sink 从 Channel 拉取数据，写入目标系统

👉 示例流程：
Web日志 -> Source（taildir） -> Channel（Memory Channel） -> Sink（HDFS）

4. Flume 部署模式

Flume 支持灵活的部署方式，主要有三种：

  • 单机模式：Source、Channel、Sink 都在同一节点，简单但不适合大规模数据
  • 多机流模式：多个 Flume 节点串联，Source 采集数据，Sink 输出到下一个 Flume 节点的 Source，逐层转发
  • 多 Agent 模式：多个 Flume Agent 独立采集数据，汇总到统一 Sink

5. 本地部署(单机)

下载地址：http://archive.apache.org/dist/flume/

tar -zxf apache-flume-1.9.0-bin.tar.gz -C /export/server/
cd /export/server/
mv apache-flume-1.9.0-bin/ flume

--将 lib 文件夹下的 guava-11.0.2.jar 删除以兼容 Hadoop 3.1.3
cd flume/lib
rm guava-11.0.2.jar

6.Flume 入门案例

6.1 监控端口数据官方案例

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

实现步骤：

yum install -y nc

--判断 44444 端口是否被占用
netstat -nlp | grep 44444

cd /export/server/flume/
mkdir job
cd job/

vim net-flume-logger.conf
--添加如下内容
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444

# 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 监听端口

第一种写法：

bin/flume-ng agent --conf conf/ --name a1 --conf-file job/net-flume-logger.conf -
Dflume.root.logger=INFO,console

第二种写法：

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

参数说明：

--conf/-c：表示配置文件存储在 conf/目录

--name/-n：表示给 agent 起名为 a1

--conf-file/-f：flume 本次启动读取的配置文件是在 job 文件夹下的 flume-telnet.conf 文件。

-Dflume.root.logger=INFO,console ：-D 表示 flume 运行时动态修改 flume.root.logger 参数属性值，并将控制台日志打印级别设置为 INFO 级别。日志级别包括:log、info、warn、 error。

此时服务端已经开启，新建一个会话作为客户端，使用 netcat 工具向本机的 44444 端口发送内容

nc localhost 44444
hello

在 Flume 监听页面观察接收数据情况

6.2 实时监控单个追加文件

案例需求：实时监控 Hive 日志，并上传到 HDFS 中

实现步骤：

确认 Hadoop 和 Hive 环境已经配置，没有配置的可以参考这两篇文章

本地部署HDFS集群https://blog.csdn.net/m0_73641796/article/details/145998092?spm=1001.2014.3001.5501


本地部署Hive集群https://blog.csdn.net/m0_73641796/article/details/146078614?spm=1001.2014.3001.5501

创建 flume-file-hdfs.conf 文件 

vim flume-file-hdfs.conf
--添加如下内容

# 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 /export/server/hive/logs/hive.log

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

# 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

确保 hadoop 用户有写入权限

mkdir -p /export/server/flume/logs
chown -R hadoop:hadoop /export/server/flume/logs
chmod -R 755 /export/server/flume/logs

 运行 Flume（用Hadoop用户运行，因为要操作HDFS） 

bin/flume-ng agent --conf conf/ --name a2 --conf-file job/flume-file-hdfs.conf

新建会话，开启 Hadoop 和 Hive 并操作 Hive 产生日志

su hadoop

start-dfs.sh
start-yarn.sh

cd /export/server/hive/
nohup bin/hive --service metastore  >> logs/metastore.log 2>&1 &
bin/hive

在 HDFS 上查看文件

6.3 实时监控目录下多个新文件

案例需求：使用 Flume 监听整个目录的文件，并上传至 HDFS

实现步骤：

创建配置文件

vim flume-dir-hdfs.conf
--添加如下内容

a3.sources = r3
a3.sinks = k3
a3.channels = c3

# Describe/configure the source
a3.sources.r3.type = spooldir
a3.sources.r3.spoolDir = /export/server/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://node1:8020/flume/upload/%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 = 100
# 设置文件类型，可支持压缩
a3.sinks.k3.hdfs.fileType = DataStream
# 多久生成一个新的文件
a3.sinks.k3.hdfs.rollInterval = 30
# 设置每个文件的滚动大小大概是 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

创建目录并启动

--创建目录
mkdir upload
chown hadoop:hadoop upload/

--启动flume（用hadoop用户）
bin/flume-ng agent --conf conf/ --name a3 --conf-file job/flume-dir-hdfs.conf

新建会话，向 upload 文件夹中添加文件

cd /export/server/flume/upload/
echo "hello world" > 1.txt

查看 HDFS 上的数据

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

案例需求:使用 Flume 监听整个目录的实时追加文件，并上传至 HDFS

实现步骤：

创建配置文件

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

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

创建目录并启动

--创建目录
mkdir files files2
chown hadoop:hadoop files
chown hadoop:hadoop files2

--启动flume（用hadoop用户）
bin/flume-ng agent --conf conf/ --name a3 --conf-file job/flume-taildir-hdfs.conf

向 upload 文件夹中添加文件

cd files
echo hello >> file1.txt

查看 HDFS 上的数据

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/2317506.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

【Linux系列】文件压缩

【Linux系列】文件压缩

💝💝💝欢迎来到我的博客，很高兴能够在这里和您见面！希望您在这里可以感受到一份轻松愉快的氛围，不仅可以获得有趣的内容和知识，也可以畅所欲言、分享您的想法和见解。 推荐:kwan 的首页,持续学…
阅读更多...
微服务架构中10个常用的设计模式

微服务架构中10个常用的设计模式

​在当今的微服务架构中，常见的十种设计模式，分别是服务发现模式、API网关模式、断路器模式、边车模式、负载均衡模式、Saga事务模式、CQRS模式、分片模式、分布式日志跟踪模式、熔断与降级模式 。其中，服务发现模式十分关键，通过…
阅读更多...
Vue3组件+leaflet，实现重叠marker的Popup切换显示

Vue3组件+leaflet，实现重叠marker的Popup切换显示

一、前言 GIS开发过程中，经常需要绘制marker，这些marker很大概率会有坐标相同导致的叠加问题，这种情况下会降低使用体验感。所以我们可以将叠加的marker的popup做一个分页效果，可以切换显示的marker。 二、技术要点 我们以leaf…
阅读更多...
机器学习之距离度量方法

机器学习之距离度量方法

常见的距离度量方法及相关函数、图示如下： 1. 欧几里得距离（Euclidean Distance） 函数公式：对于两个 ( n ) 维向量 ( x = ( x 1 , x 2 , ⋯   ,
阅读更多...
3.1 在VisionPro脚本中添加CogGraphicLabel

3.1 在VisionPro脚本中添加CogGraphicLabel

本案例需要实现如下功能： 1.加载toolBlock 2.加载图片， 3.运行Block 4.VisionPro中添加脚本显示数值。 见下图：详细代码（C#以及visionPro）见下面链接： https://download.csdn.net/download/qq_340474…
阅读更多...
AI：Machine Learning Data Science

AI：Machine Learning Data Science

机器学习与数据科学 左侧 机器学习 Machine Learning 机器学习是一门多领域交叉学科，涉及概率论、统计学、逼近论、凸分析、算法复杂度理论等多门学科。专门研究计算机怎样模拟或实现人类的学习行为，以获取新的知识或技能，重新组织已有的知…
阅读更多...
软件需求分类、需求获取（高软46）

软件需求分类、需求获取（高软46）

系列文章目录 软件需求分类，需求获取 文章目录 系列文章目录前言一、软件需求二、获取需求三、真题总结 前言 本节讲明软件需求分类、需求获取的相关知识。 一、软件需求 二、获取需求 三、真题 总结 就是高软笔记，大佬请略过！
阅读更多...
嵌入式Linux | 什么是 BootLoader、Linux 内核（kernel）、和文件系统?

嵌入式Linux | 什么是 BootLoader、Linux 内核（kernel）、和文件系统?

01 什么是 BootLoader 呢？ 它是个引导程序，也就是硬件复位以后第一个要执行的程序，它主要工作就是初始化操作系统运行的环境，比如说内存、定时器、缓冲器等，当这个工作做完以后，再把操作系统的代码加载…
阅读更多...
函数（函数的概念、库函数、自定义函数、形参和实参、return语句、数组做函数参数、嵌套调用和链式访问、函数的声明和定义、static和extern）

函数（函数的概念、库函数、自定义函数、形参和实参、return语句、数组做函数参数、嵌套调用和链式访问、函数的声明和定义、static和extern）

一、函数的概念 •C语⾔中的函数：⼀个完成某项特定的任务的⼀⼩段代码 •函数又被翻译为子函数（更准确） •在C语⾔中我们⼀般会⻅到两类函数：库函数 ⾃定义函数 二、库函数 1 .标准库和头文件 •C语⾔的国际标准ANSIC规定了⼀…
阅读更多...
ImGui 学习笔记（五） —— 字体文件加载问题

ImGui 学习笔记（五） —— 字体文件加载问题

ImGui 加载字体文件的函数似乎存在编码问题，这一点可能跟源文件的编码也有关系，我目前源文件编码是 UTF-16。 当参数中包含中文字符时，ImGui 内部将字符转换为宽字符字符集时候，采用的 MultiByteToWideChar API 参数不太对&#…
阅读更多...
OpenCV计算摄影学（20）非真实感渲染之增强图像的细节函数detailEnhance()

OpenCV计算摄影学（20）非真实感渲染之增强图像的细节函数detailEnhance()

操作系统：ubuntu22.04 OpenCV版本：OpenCV4.9 IDE:Visual Studio Code 编程语言：C11 算法描述 此滤波器增强特定图像的细节。 cv::detailEnhance用于增强图像的细节，通过结合空间域和频率域的处理，提升图像中特定细节…
阅读更多...
Android PC 要来了？Android 16 Beta3 出现 Enable desktop experience features 选项

Android PC 要来了？Android 16 Beta3 出现 Enable desktop experience features 选项

在之前的 《Android 桌面窗口新功能推进》 我们就聊过，Google 就一直在努力改进 Android 的内置桌面模式，例如添加了适当的窗口标题、捕捉窗口的能力、悬停选项、窗口大小调整、最小化支持、app-to-web 等。 比如在搭载 Android 15 QPR 1 Beta 2 的 Pix…
阅读更多...
Git常用操作之GitLab

Git常用操作之GitLab

Git常用操作之GitLab 小薛博客官网：小薛博客Git常用操作之GitLab官方地址 1、GitLab安装 https://gitlab.cn/install/ 1、Docker安装GitLab https://docs.gitlab.cn/jh/install/docker.html 1、设置卷位置 在设置其他所有内容之前，请配置一个新的…
阅读更多...
Netty基础—NIO的使用简介

Netty基础—NIO的使用简介

1.Buffer缓冲区 (1)Buffer缓冲区的作用 在NIO中，所有的数据都是通过使用Buffer缓冲区来处理的。如果要通过NIO，将数据写到文件和网络或从文件和网络中读取数据，那么就需要使用Buffer缓冲区来进行处理。 (2)Buffer缓冲区的4个核心概念 Buffer缓…
阅读更多...
Matlab 汽车ABS实现模糊pid和pid控制

Matlab 汽车ABS实现模糊pid和pid控制

1、内容简介 Matlab 181-汽车ABS实现模糊pid和pid控制 可以交流、咨询、答疑 2、内容说明 略 实现汽车防抱死制动系统（ABS）的控制算法，通常涉及到传统的PID控制和模糊PID控制两种方法。下面将分别介绍这两种控制策略的基本概念以及如何在M…
阅读更多...
Muon: An optimizer for hidden layers in neural networks

Muon: An optimizer for hidden layers in neural networks

引言 在深度学习领域，优化算法对模型训练效率和性能起着关键作用。从经典的随机梯度下降 (SGD) 及其动量法，到自适应优化方法 Adam/AdamW 等，一系列优化器大大加速了神经网络的收敛。然而，随着模型规模和数据量的爆炸式增长&…
阅读更多...
【VSCODE 插件 可视化】：SVG 编辑插件 SVG Editor

【VSCODE 插件 可视化】：SVG 编辑插件 SVG Editor

插件下载 svgeditor 创建文件 Windows/Linux 快捷键 Ctrl Shift P 打开VSCODE 命令面板查找 New File With Svg Editor 编辑文件 保存文件 打开文件以继续编辑 CG 选中多个：shift单击没找到横向分布功能无法用键盘微调位置
阅读更多...
Cursor插件市场打不开解决

Cursor插件市场打不开解决

问题现象： cursor搜索插件的时候提示错误，无法搜索安装插件 error while fetching extensions.failed to fetch 问题原因 cursor默认安装使用的并不是vs code的插件市场，国内网络有时候打不开 解决 修改插件市场地址并重启cursor 打开cur…
阅读更多...
嵌入式开发之STM32学习笔记day06

嵌入式开发之STM32学习笔记day06

基于STM32F103C8T6的开发实践——从入门到精通01 1. 引言 STM32系列微控制器是STMicroelectronics推出的一款高性能、低功耗的32位微控制器，广泛应用于嵌入式系统中。STM32F103C8T6是其中非常受欢迎的一款，凭借其强大的性能、丰富的外设接口和低廉的价格…
阅读更多...
AI驱动的视频字幕提取与翻译工具

AI驱动的视频字幕提取与翻译工具

青梧字幕是一款基于Whisper技术的AI字幕提取工具，专为视频制作者、翻译人员和自媒体创作者设计。它通过先进的语音识别算法，能够自动从视频文件中提取字幕内容，并支持多种语言和字幕格式，极大地简化了字幕制作流程。 目前暂支持 …
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023