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

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
检索分析服务 Elasticsearch 版,2核4GB开发者规格 1个月
大数据开发治理平台 DataWorks,不限时长
简介: Apache Flume 是一个开源的数据采集工具,用于从各种数据源(如日志、网络数据、消息队列)收集大规模数据,并将其传输和加载到数据存储系统(如 HDFS、HBase、Hive)。Flume 由数据源(Source)、通道(Channel)、拦截器(Interceptor)和接收器(Sink)组成,支持灵活配置以适应不同的数据流处理需求。安装 Flume 包括解压软件包、配置环境变量和调整日志及内存设置。配置文件定义数据源、通道、拦截器和接收器,拦截器允许预处理数据。Flume 适用于构建数据管道,整合分散数据到中心存储系统,便于分析和报告。

@[toc]

什么是 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

image.png

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

# 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-ngflume-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¶),如下所示:

image.png

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

拦截器应用

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

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

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





相关文章
|
1月前
|
消息中间件 存储 分布式计算
【Flume 】Flume 基础概述
【4月更文挑战第4天】【Flume 】Flume 基础概述
|
1月前
|
消息中间件 分布式计算 大数据
【大数据技术Hadoop+Spark】Flume、Kafka的简介及安装(图文解释 超详细)
【大数据技术Hadoop+Spark】Flume、Kafka的简介及安装(图文解释 超详细)
126 0
|
9月前
|
SQL 分布式计算 监控
大数据Flume快速入门
大数据Flume快速入门
65 0
|
24天前
|
存储 数据采集 监控
Flume 拦截器概念及自定义拦截器的运用
Apache Flume 的拦截器是事件处理组件,位于Source和Channel之间,用于在写入Channel前对数据进行转换、提取或删除。它们支持数据处理和转换、数据增强、数据过滤以及监控和日志功能。要创建自定义拦截器,需实现Interceptor接口,包含initialize、intercept、intercept(List&lt;Event&gt;)和close方法。配置拦截器时,通过Builder模式实现Interceptor.Builder接口。在Flume配置文件中指定拦截器全类名,如`TestInterceptor$Builder`,然后启动Flume进行测试。
|
1月前
|
存储 监控 Linux
Flume【部署 02】Flume监控工具Ganglia的安装与配置(CentOS 7.5 在线安装系统监控工具Ganglia + 权限问题处理 + Flume接入监控配置 + 图例说明)
【2月更文挑战第17天】Flume【部署 02】Flume监控工具Ganglia的安装与配置(CentOS 7.5 在线安装系统监控工具Ganglia + 权限问题处理 + Flume接入监控配置 + 图例说明)
75 1
Flume【部署 02】Flume监控工具Ganglia的安装与配置(CentOS 7.5 在线安装系统监控工具Ganglia + 权限问题处理 + Flume接入监控配置 + 图例说明)
|
10月前
|
SQL 分布式计算 监控
Flume学习--1、Flume概述、Flume入门、(一)
Flume学习--1、Flume概述、Flume入门、(一)
|
1月前
|
存储 监控 Linux
Ganglia【部署 01】Flume监控工具Ganglia的安装与配置(CentOS 7.5 在线安装系统监控工具Ganglia + 权限问题处理 + Flume接入监控配置 + 图例说明)
Ganglia【部署 01】Flume监控工具Ganglia的安装与配置(CentOS 7.5 在线安装系统监控工具Ganglia + 权限问题处理 + Flume接入监控配置 + 图例说明)
68 0
|
1月前
|
数据采集 消息中间件 缓存
Apache Flume及快速安装
Apache Flume及快速安装
50 0
|
1月前
|
存储 分布式计算 监控
Flume(一)【Flume 概述】
Flume(一)【Flume 概述】
|
1月前
|
消息中间件 存储 分布式计算
Hadoop学习笔记(HDP)-Part.20 安装Flume
01 关于HDP 02 核心组件原理 03 资源规划 04 基础环境配置 05 Yum源配置 06 安装OracleJDK 07 安装MySQL 08 部署Ambari集群 09 安装OpenLDAP 10 创建集群 11 安装Kerberos 12 安装HDFS 13 安装Ranger 14 安装YARN+MR 15 安装HIVE 16 安装HBase 17 安装Spark2 18 安装Flink 19 安装Kafka 20 安装Flume
70 0
Hadoop学习笔记(HDP)-Part.20 安装Flume

热门文章

最新文章