Flume 快速入门【概述、安装、拦截器】

文章目录

    • 什么是 Flume？
    • Flume 组成
    • Flume 安装
    • Flume 配置任务文件
      • 应用示例
      • 启动 Flume 采集任务
    • Flume 拦截器
      • 编写 Flume 拦截器
      • 拦截器应用

什么是 Flume？

Flume 是一个开源的数据采集工具，最初由 Apache 软件基金会开发和维护。它的主要目的是帮助用户将大规模数据从各种数据源（如日志文件、网络数据源、消息队列等）采集、传输和加载到数据存储系统（如 Hadoop HDFS、Apache HBase、Apache Hive 等）。

Flume 旨在处理大规模数据流，以便进行数据分析和处理。

Flume 组成

Flume （配置）主要由以下 4 个部分组成：

1. 数据源（Source）： Flume 可以从多种数据源收集数据，例如日志文件、网络流、消息队列等。

2. 通道（Channel）： 采集的数据被存储在通道中，等待传输到目标数据存储系统。Flume 支持多种不同类型的通道，如内存通道、文件通道和 Kafka 通道。

3. 拦截器（Interceptor）： 拦截器允许用户对采集的数据进行预处理和转换，以满足特定需求。

4. 接收器（Sink）： 接收器将数据传输到目标数据存储系统，如 Hadoop HDFS、HBase、Kafka 等。

Flume 通过灵活的配置，允许用户根据其数据采集需求来定义数据流的整个流程，包括数据源、通道、拦截器和接收器。

这使得 Flume 成为处理大规模数据采集和传输任务的强有力工具，构建数据管道，将分散的数据整合到中心存储或处理系统中，用于实时或者离线数据分析和报告。

Flume 安装

官方安装包下载地址：http://archive.apache.org/dist/flume

本篇博客使用的版本为：Flume-1.10.1

1. 解压

tar -zxvf apache-flume-1.10.1-bin.tar.gz -C /opt/

2. 配置环境变量

vim /etc/profile

文件末尾添加：

#FLUME_HOME
export FLUME_HOME=/opt/flume-1.10.1
export PATH=$PATH:$FLUME_HOME/bin

刷新环境变量：source /etc/profile

其实到这里，Flume 算是安装完了，但是为了后期使用方便，这里再调整一下配置参数。

修改日志存储与输出：

cd $FLUME_HOME

vim conf/log4j2.xml

在该文件中修改日志文件的存储目录（正文第 3 行）

    <Property name="LOG_DIR">/opt/flume-1.10.1/logs</Property>

在该文件中添加日志控制台输出方式（正文末尾）

      <AppenderRef ref="Console" />

默认只有 LogFile 日志文件的输出方式。


修改堆内存大小：

cd $FLUME_HOME

vim conf/flume-env.sh

如果是本地学习或者测试环境建议调小一点：

export JAVA_OPTS="-Xms512m -Xmx2048m -Dcom.sun.management.jmxremote"

我这里调整最小为 512MB，最大 2048MB，也可以将最大和最小调整为一样的，避免进行内存交换。

Flume 配置任务文件

Flume 最主要的内容就是配置任务文件了，在文章开头提到过，主要由四部分组成：

  • 数据源（Source）

  • 通道（Channel）

  • 拦截器（Interceptor）

  • 接收器（Sink）

我们可以根据需求，进入 Flume 的官方网站，查阅各项参数如何进行配置，按照要求配置即可。

配置查阅网站：Flume 1.10.1 User Guide

在这里插入图片描述

其中给出了一个模板文件，内容如下所示：

# example.conf: A single-node Flume configuration

# 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

根据该模板文件就可以来快速构建一个数据采集的配置文件啦。

应用示例

将 Maxwell 发送到 Kafka 消息队列中的数据采集到 HDFS 上。

# --------- 声明变量名称 ---------
a1.sources = r1
a1.sinks = k1
a1.channels = c1

# --------- 配置数据源 ---------
# 指定数据源类型
a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
# 指定连接地址
a1.sources.r1.kafka.bootstrap.servers = hadoop120:9092,hadoop121:9092,hadoop122:9092
# 指定消费者组别，防止多个消费者之间引发数据冲突
a1.sources.r1.kafka.consumer.group.id = flume1
# 指定主题名称，这里需要和 MaxWell 指定发送的主题保持一致，否则会采集不到数据
a1.sources.r1.kafka.topics = topic_db

# --------- 配置接收器 ---------
# 指定存储源类型
a1.sinks.k1.type = hdfs
# 动态规划 HDFS 写入路径
a1.sinks.k1.hdfs.path = /test/%{tableName}_inc/%Y/%m/%d/

# 当以下值中的其中一个满足时，触发滚动操作，将数据写入到新的文件中（避免小文件过多）
# 根据运行时间判定（s），测试环境调小，开发环境30m-1h
a1.sinks.k1.hdfs.rollInterval = 10 
# 根据数据量大小判定（B），128 MB
a1.sinks.k1.hdfs.rollSize = 134217728
# 根据文件的条数判断，为 0 时表示不依据该参数
a1.sinks.k1.hdfs.rollCount = 0

# 压缩文件
# 指定文件类型为压缩流
a1.sinks.k1.hdfs.fileType = CompressedStream
# 指定数据压缩格式
a1.sinks.k1.hdfs.codeC = gzip

# --------- 配置通道 ---------
# 通道类型
a1.channels.c1.type = file
# 检查点存储路径
a1.channels.c1.checkpointDir = /opt/module/flume-1.10.1/file-channel/checkpoint1
# 用于存储日志文件的目录，多个路径用逗号分隔
a1.channels.c1.dataDirs = /opt/module/flume-1.10.1/file-channel/data1
# 指定允许等待的时间
a1.channels.c1.keep-alive = 6


# --------- 组装 ---------
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

启动 Flume 采集任务

cd $FLUME_HOME

./bin/flume-ng agent -c conf/ -f job_file -n a1

参数解析：

  • ./bin/flume-ng：flume-ng 是 Flume 的执行脚本，它用于启动 Flume 的 agent 实例。

  1. agent：这告诉 Flume-ng 启动一个代理实例，也就是一个数据采集和传输任务的执行单元。

  2. -c conf/：指定 Flume 配置文件的目录。

  3. -f job_file：指定 Flume 任务配置文件的参数，其中配置文件包含了数据源、通道、接收器以及数据处理的详细信息。

  4. -n a1：指定代理实例的名称，与配置文件中的对应。

Flume 拦截器

当我们在配置文件中定义了动态参数时，例如上方示例中接收器的配置语句：

a1.sinks.k1.hdfs.path = /test/%{tableName}_inc/%Y/%m/%d/

我们设想的是将表名称和年月日进行动态规划，但在未设置拦截器时，这些动态参数值都会被默认为空，如果是系统预定义的参数则为系统设定值。

如下所示：


其中 tableName 是自定义的值，Flume 系统并没有对其进行预定义，所以为空，但 %Y %m %d 这三个值系统默认为当前的日期值，所以不为空。

如果想将上述值设定为希望出现的值，此时便引出了拦截器的概念。通过对拦截器的配置，将采集的数据进行预处理和转换，以满足特定需求。

编写 Flume 拦截器

在 IDEA 中编写拦截器代码，然后打包上传，使用依赖如下所示：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.work</groupId>
    <artifactId>intercepted</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <dependencies>

        <!-- JSON 解析包-->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>2.0.32</version>
        </dependency>

        <!-- flume 包，不打包该 Jar 包-->
        <dependency>
            <groupId>org.apache.flume</groupId>
            <artifactId>flume-ng-core</artifactId>
            <version>1.10.1</version>
            <scope>provided</scope>
        </dependency>

    </dependencies>
    
    <build>
        <plugins>
            <plugin>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>2.3.2</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
            <plugin>
                <artifactId>maven-assembly-plugin</artifactId>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

</project>

注意，JDK 版本与平台保持一致。

拦截器实现，用于设定表头与写入日期：

package com.work.flume.interceptor;

import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;

/**
 * @author Moon_coder
 * @version 1.0
 * @date 2023/10/29 17:32
 */
public class TableNameAndTimestamp implements Interceptor {

    /**
     * 初始化方法
     */
    @Override
    public void initialize() {

    }

    /***
     * 处理单条数据
     * @param event
     * @return
     */
    @Override
    public Event intercept(Event event) {
        try{
            // 1.获取头数据
            Map<String, String> headers = event.getHeaders();
            // 2.获取数据内容，将字节数据转换为字符串
            String log = new String(event.getBody(), StandardCharsets.UTF_8);
            // 3.将字符串转换为 JSON 对象
            JSONObject jsonObject = JSONObject.parseObject(log);
            // 4.获取表名
            String table = jsonObject.getString("table");
            // 5.获取时间，我的数据是经 Maxwell 采集的，Maxwell 中的数据是 10 位时间戳，不含毫秒，将数据存入 HDFS 时 *1000
            String ts = jsonObject.getString("ts") + "000";
            // 6.更新头数据信息
            headers.put("tableName",table);
            headers.put("timestamp",ts);
        }catch (JSONException e){
            // 如果不是 JSON 数据，则将该数据定义为脏数据
            return null;
        }
        return event;
    }

    /***
     * 批量数据处理方法
     * @param events
     * @return
     */
    @Override
    public List<Event> intercept(List<Event> events) {
        // 批量处理 event 同时实现过滤功能
        events.removeIf(next -> intercept(next) == null);

        return events;
    }

    /**
     * 关闭方法
     */
    @Override
    public void close() {

    }

    // TODO 返回拦截器类
    public static class Builder implements Interceptor.Builder{

        @Override
        public Interceptor build() {
            return new TableNameAndTimestamp();
        }

        @Override
        public void configure(Context context) {

        }
    }


}

类名或 Jar 包名称都没有特别要求，自定义即可。

注意： 当我们在往头信息里面放东西时，需要与键名一一对应。

            // 6.更新头数据信息
            headers.put("tableName",table);
            headers.put("timestamp",ts);

如果是自定义的值，名称与 Flume 配置文件设定的必须对应：


如果是系统预定义的值，则需要在官方网站中查询其对应的键名。例如这里出现的 %Y %m %d 这三个值，在接收器的参数定义那里即可查询到（HDFS Sink¶），如下所示：

在这里插入图片描述

这里官方给出了提示，说 对于所有与时间相关的转义序列，事件的标头中必须存在一个关键字为 “timestamp” 的标头，所以在拦截器中对头信息的时间进行操作时，对应的键名为 timestamp。

拦截器应用

将打包好的拦截器 Jar 包上传到 Flume 中的 lib 目录下，然后在 Flume 任务配置文件中添加拦截器配置，如下所示：

# --------- 拦截器 ---------
# 拦截器名称
a1.sources.r1.interceptors = i1
# 编写的拦截器全类名 + $Builder 标识符
a1.sources.r1.interceptors.i1.type = com.work.flume.interceptor.TableNameAndTimestamp$Builder

再次执行上方的示例任务，可以看到配置完拦截器后，头信息已经达到了我们预期的结果。


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

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

9、电路综合-基于简化实频的任意幅频响应的微带电路设计

9、电路综合-基于简化实频的任意幅频响应的微带电路设计

9、电路综合-基于简化实频的任意幅频响应的微带电路设计 网络综合和简化实频理论学习概述中的1-8介绍了SRFT的一些基本概念和实验方法&#xff0c;终于走到了SRFT的究极用途&#xff0c;给定任意响应直接综合出微带电路。 1、任意幅频响应的微带电路设计用途 我们演示了采用…
阅读更多...
Python皮卡丘

Python皮卡丘

系列文章 序号文章目录直达链接1浪漫520表白代码https://want595.blog.csdn.net/article/details/1306668812满屏表白代码https://want595.blog.csdn.net/article/details/1297945183跳动的爱心https://want595.blog.csdn.net/article/details/1295031234漂浮爱心https://want…
阅读更多...
[C++]命名空间等——喵喵要吃C嘎嘎

[C++]命名空间等——喵喵要吃C嘎嘎

希望你开心&#xff0c;希望你健康&#xff0c;希望你幸福&#xff0c;希望你点赞&#xff01; 最后的最后&#xff0c;关注喵&#xff0c;关注喵&#xff0c;关注喵&#xff0c;大大会看到更多有趣的博客哦&#xff01;&#xff01;&#xff01; 喵喵喵&#xff0c;你对我真的…
阅读更多...
Python爬虫实战（六）——使用代理IP批量下载高清小姐姐图片（附上完整源码）

Python爬虫实战（六）——使用代理IP批量下载高清小姐姐图片（附上完整源码）

文章目录 一、爬取目标二、实现效果三、准备工作四、代理IP4.1 代理IP是什么&#xff1f;4.2 代理IP的好处&#xff1f;4.3 获取代理IP4.4 Python获取代理IP 五、代理实战5.1 导入模块5.2 设置翻页5.3 获取图片链接5.4 下载图片5.5 调用主函数5.6 完整源码5.7 免费代理不够用怎…
阅读更多...
C++ 自引用指针this（整理）

C++ 自引用指针this（整理）

使用例子&#xff1a; #include <iostream> #include <Windows.h> using namespace std; class A { public:A(int x1){x x1;}void disp(){cout<<"this"<<this<<" when x"<<this->x<<endl;} private:int x;…
阅读更多...
【工具使用】NPS内网穿透工具介绍

【工具使用】NPS内网穿透工具介绍

文章目录 前言一、内网穿透二、NPS概述三、NPS原理四、NPS服务器搭建(一)云服务器配置 五、NPS内网穿透演示(一)演示案例一(二)演示案例二 六、NPS内网穿透检测建议(一)流量监控(二)流量协议分析(三)网络行为异常检测 七、NPS内网穿透防范建议(一)阻止或隔离流量(二)更新和强化…
阅读更多...
14技术太卷我学APEX-RESTful服务全自动Autorest

14技术太卷我学APEX-RESTful服务全自动Autorest

Oracle REST 数据服务 (ORDS) 的 AutoREST 功能&#xff0c;该功能允许您将表和视图等自动公开为 RESTful Web 服务。这个太好可不用自己费劲巴拉的定义RESTful数据服务了。 1 先准备测试数据表 登录APEX现在开发平台 [https://apex.oracle.com/pls/apex/r/apex/workspace-si…
阅读更多...
Unity UniTask【安装】

Unity UniTask【安装】

窗口→包管理器&#xff0c;输入GitURL&#xff1a;https://github.com/Cysharp/UniTask.git?pathsrc/UniTask/Assets/Plugins/UniTask 如果上述方式不行&#xff0c;那就直接导入UniTask包&#xff0c;下载地址&#xff1a;Unity3D异步神器UniTask2.4.1包资源-CSDN文库
阅读更多...
DBA笔记（1）

DBA笔记（1）

目录 1、rpm yum 命令的使用&#xff0c;参数的含义 rpm命令&#xff1a; yum命令&#xff1a; 2、上传镜像至虚拟机搭建本地yum源 3、chown chomd 命令每一个参数的含义 chown命令&#xff1a; chmod命令&#xff1a; 4、fdisk partd 硬盘分区命令用法 fdisk命令&am…
阅读更多...
Seata入门系列【15】@GlobalLock注解使用场景及源码分析

Seata入门系列【15】@GlobalLock注解使用场景及源码分析

1 前言 在Seata 中提供了一个全局锁注解GlobalLock&#xff0c;字面意思是全局锁&#xff0c;搜索相关文档&#xff0c;发现资料很少&#xff0c;所以分析下它的应用场景和基本原理&#xff0c;首先看下源码中对该注解的说明&#xff1a; // 声明事务仅在单个本地RM中执行 //…
阅读更多...
【错误解决方案】ModuleNotFoundError: No module named ‘ngboost‘

【错误解决方案】ModuleNotFoundError: No module named ‘ngboost‘

1. 错误提示 在python程序&#xff0c;尝试导入一个名为ngboost的模块&#xff0c;但Python提示找不到这个模块。 错误提示&#xff1a;ModuleNotFoundError: No module named ‘ngboost‘ 2. 解决方案 出现上述问题&#xff0c;可能是因为你还没有安装这个模块&#xff0c;…
阅读更多...
探索环幕投影系统在沉浸式展厅设计中的应用优势

探索环幕投影系统在沉浸式展厅设计中的应用优势

伴随着数字化时代的快速发展&#xff0c;以往的静态展示方式已经无法满足人们对内容展览展示的需求&#xff0c;因此需要利用先进的多媒体互动技术&#xff0c;来提升观众的参观体验&#xff0c;而其中环幕投影系统便是用于建设沉浸式空间的主流技术之一&#xff0c;其被应用在…
阅读更多...
【Linux】Linux的安装以及常见命令

【Linux】Linux的安装以及常见命令

&#x1f973;&#x1f973;Welcome Huihuis Code World ! !&#x1f973;&#x1f973; 接下来看看由辉辉所写的关于Linux的相关操作吧 一.Linux的安装 1.创建虚拟机 2.选择linux 3.配置虚拟机 4.开启虚拟机 默认回车即可 5.安装linux 5.登录账户 6.解决网络问题 ①先查看一下…
阅读更多...
宇信科技：强势行业加速融入AIGC，同时做深做细

宇信科技：强势行业加速融入AIGC，同时做深做细

【科技明说 &#xff5c; 重磅专题】 大家可能没有想到&#xff0c;一向对外低调行事的宇信科技&#xff0c;在AIGC方面2023年就已经训练出了适配金融场景的垂直模型&#xff0c;并应用到了各产品线上&#xff0c;同时结合通用大模型预研了宇信金融系统编程大模型。宇信金融系…
阅读更多...
IDEA优雅自动生成类注释和快捷键生成方法注释

IDEA优雅自动生成类注释和快捷键生成方法注释

生成类注释 Preferences->Editor->File and Code Templates-> Includes ->File Header 注释模板&#xff1a; /*** Classname ${NAME}* Description ${description}* Date ${DATE} ${TIME}* Created by ZouLiPing*/生成方法和字段注释 查看IDEA自动配置java快捷…
阅读更多...
Qwt QwtScaleDraw自定义坐标轴

Qwt QwtScaleDraw自定义坐标轴

1.概述 QwtScaleDraw 是 Qt 绘图库 Qwt 中的一个类&#xff0c;用于绘制坐标轴刻度线和刻度标签。它提供了一些方法和属性来设置刻度线和标签的样式、布局和对齐方式。 以下是类继承关系&#xff1a; 2.常用方法 标签相关方法&#xff1a; setLabelRotation(double angle)&…
阅读更多...
UE5关键字查询功能

UE5关键字查询功能

阅读更多...
一、【Photoshop如何根据不同类型图像抠图】

一、【Photoshop如何根据不同类型图像抠图】

文章目录 前言图形结构1、规则图形2、不规则图形 图形颜色1、轮廓清晰2、颜色分明 前言 当我们有抠图需求的时候&#xff0c;不要一开始就想着我怎么去把它抠出来&#xff0c;首先应该分析图形的特点&#xff0c;然后再去选取合适的工具&#xff0c;这样才可以做到事半功倍&am…
阅读更多...
UDP数据报套接字编程

UDP数据报套接字编程

1. 什么是网络编程&#xff1f; 网络编程&#xff0c;指网络上的主机&#xff0c;通过不同的进程&#xff0c;以编程的方式实现网络通信&#xff08;或称为网络数据传输&#xff09;。 这里只是在开发学习的时候使用的是&#xff0c;不同的进程来代表不同的主机来模拟网络通信…
阅读更多...
python：使用Scikit-image对遥感影像进行傅里叶变换特征提取（fourier）

python：使用Scikit-image对遥感影像进行傅里叶变换特征提取（fourier）

作者：CSDN @ _养乐多_ 在本博客中，我们将介绍如何使用Scikit-Image来进行傅里叶变换特征提取（fourier），并且提供一个示例代码，演示了如何在单波段遥感图像上应用这些方法。 傅里叶变换特征提取是一种数学工具，用于将图像中的细节、纹理和边缘信息以不同频率的方式呈现…
阅读更多...
最新文章

Copyright @ 2022~2023