文章目录
- Flume 拦截器
- 拦截器的作用
- 拦截器运用
- 1.创建项目
- 2.实现拦截器接口
- 3.编写事件处理逻辑
- 4.拦截器构建
- 5.打包与上传
- 6.编写配置文件
- 7.测试运行
Flume 拦截器
在 Flume 中,拦截器(Interceptors)是一种可以在事件传输过程中拦截、处理和修改事件的组件。
位于 Source 与 Channel 之间,在写入Channel 之前,拦截器可以对数据进行转换、提取或删除,以满足特定的需求。每个拦截器只处理同一个 Source 接收到的事件,你也可以同时配置多个拦截器,它们会按顺序执行。
拦截器的作用
-
数据处理和转换: 拦截器可以对事件数据进行处理和转换。例如,可以对原始日志进行解析、过滤、格式化等操作,以便后续处理或存储。
-
数据增强: 拦截器可以为事件数据添加额外的信息或元数据。例如,可以添加时间戳、主机信息、标签等,以丰富事件数据的内容。
-
数据过滤: 拦截器可以根据特定条件过滤掉不需要的事件数据,减少数据传输的量或过滤掉无效数据。
-
监控和日志: 拦截器可以用于监控数据流的运行情况,记录日志信息或统计数据流中的事件数量、处理速率等指标,帮助用户进行性能分析和故障排查。
拦截器运用
1.创建项目
创建一个 Maven 工程项目,引入 Flume 依赖。
在 IDEA 中创建 Maven 项目想必大家都会,这里不再赘述。
根据集群中的 Flume 版本,引入 Flume 依赖,如下所示:
<dependencies>
<dependency>
<groupId>org.apache.flume</groupId>
<artifactId>flume-ng-core</artifactId>
<version>1.10.1</version>
<scope>provided</scope>
</dependency>
</dependencies>
无需将该依赖打包进最后的 JAR 包中,故将其作用域设置为 provided
。
当一个依赖项的 scope 被设置为
compile
时,它将在编译和运行时都可用,并包含在最终的项目包中。而provided
范围的依赖项仅在编译和测试阶段需要,运行时不包括。
2.实现拦截器接口
创建测试类 TestInterceptor
实现拦截器 Interceptor
,注意,导包时不要导错了。
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import java.util.List;
public class TimestampInterceptor implements Interceptor {
@Override
public void initialize() {
}
@Override
public Event intercept(Event event) {
return null;
}
@Override
public List<Event> intercept(List<Event> list) {
return null;
}
@Override
public void close() {
}
}
在上面的代码中,我们实现了 Flume 拦截器接口 Interceptor
,并重写了其中的四个方法:
-
initialize()
方法:初始化拦截器操作,读取配置信息、建立连接等。 -
intercept(Event event)
方法:用于拦截单个事件,并对事件进行处理。接收一个事件对象作为输入,并返回一个修改后的事件对象。 -
intercept(List<Event> list)
方法:事件批处理,拦截事件列表,并对事件列表进行处理。 -
close()
方法:关闭拦截器,在这里释放资源、关闭连接等。
3.编写事件处理逻辑
在这里做个简单的事件处理,如果数据中包含字符串 hello
则进行过滤操作,这样我们可以直观感受到拦截器的存在,下面来进行逻辑设计。
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
public class TestInterceptor implements Interceptor {
@Override
public void initialize() {
}
@Override
public Event intercept(Event event) {
// 获取事件数据
String eventData = new String(event.getBody(), StandardCharsets.UTF_8);
// 检查事件数据中是否包含指定字符串
if (eventData.contains("hello")) {
// 如果包含指定字符串,则过滤掉该事件,返回 null
return null;
}
return event;
}
@Override
public List<Event> intercept(List<Event> events) {
// 创建一个新的列表,存储处理过后的事件
List<Event> interceptedEvents = new ArrayList<>();
for (Event event : events) {
Event interceptedEvent = intercept(event);
if (interceptedEvent != null) {
interceptedEvents.add(interceptedEvent);
}
}
return interceptedEvents;
}
@Override
public void close() {
}
}
intercept(List<Event> events)
方法用于对事件列表进行批量处理。这个方法会遍历传入的事件列表,并对每一个事件调用 intercept(Event event)
方法来进行单独处理。
注意,如果只有 intercept(Event event)
方法被重写了,而没有实现 intercept(List<Event> events)
批处理方法,那么在处理事件时会以单个事件的方式进行处理。
在不需要进行初始化和释放资源的情况下,我们可以选择不重写 initialize
和 close
方法。
4.拦截器构建
在编写完事件处理逻辑后,我们还需要对拦截器进行构建。
在 Flume 中,拦截器的创建和配置通常是通过 Builder 模式来完成的。
在程序中,我们可以定义一个静态内部类 Builder,实现 Interceptor.Builder
接口来对拦截器进行构建,如下所示:
public static class Builder implements Interceptor.Builder {
@Override
public void configure(Context context) {
// 配置操作,可留空
}
@Override
public Interceptor build() {
// 返回构建的拦截器类
}
}
在我们这个案例中,完整的代码如下所示:
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
public class TestInterceptor implements Interceptor {
@Override
public void initialize() {
}
@Override
public Event intercept(Event event) {
// 获取事件数据
String eventData = new String(event.getBody(), StandardCharsets.UTF_8);
// 检查事件数据中是否包含指定字符串
if (eventData.contains("hello")) {
// 如果包含指定字符串,则过滤掉该事件,返回 null
return null;
}
return event;
}
@Override
public List<Event> intercept(List<Event> events) {
List<Event> interceptedEvents = new ArrayList<>();
for (Event event : events) {
Event interceptedEvent = intercept(event);
if (interceptedEvent != null) {
interceptedEvents.add(interceptedEvent);
}
}
return interceptedEvents;
}
@Override
public void close() {
}
// 拦截器构建
public static class Builder implements Interceptor.Builder {
@Override
public void configure(Context context) {
}
@Override
public Interceptor build() {
return new TestInterceptor();
}
}
}
5.打包与上传
将写好的项目进行打包,并上传到集群中,进行测试。
注意,需要将打包好的拦截器包放在 Flume 安装目录下的 lib
文件夹中。
6.编写配置文件
这里为了验证拦截器的作用,通过一个 Flume 采集案例来进行体现。
如果你不知道如何编写配置文件,可以看我写的这篇文章 —— Flume 配置文件编写技巧(包会的,抄就完了)
这个配置案例是将发送到 HTTP 源中的数据采集到 HDFS 上,将本地文件作为缓冲通道,该配置文件命名为 httpToHDFS.conf
。
# 声明
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Source 源配置
a1.sources.r1.type = http
a1.sources.r1.port = 5140
a1.sources.r1.bind = localhost
# 拦截器配置
# 拦截器定义
a1.sources.r1.interceptors = i1
# 拦截器全类名
a1.sources.r1.interceptors.i1.type = TestInterceptor$Builder
# Sink 处理/存储配置
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /flume/events/%Y-%m-%d/%H%M/%S
a1.sinks.k1.hdfs.filePrefix = events-
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute
a1.sinks.k1.hdfs.useLocalTimeStamp = true
a1.sinks.k1.hdfs.fileType = CompressedStream
a1.sinks.k1.hdfs.codeC = gzip
# Channel 通道配置
a1.channels.c1.type = file
a1.channels.c1.checkpointDir = /opt/software/flume/checkpoint
a1.channels.c1.dataDirs = /opt/software/flume/data
# 组装/绑定
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
拦截器全类名配置那里需要注意,格式为 拦截器的全类名 + $Builder
。
在 IDEA 中获取全类名的方式:右击需要引用的类,依次选择【File——>Copy Path/Reference…——>Copy Reference】即可复制。
你可以根据你的需要对该配置文件进行修改。
7.测试运行
因为我们是将数据采集到 HDFS 上,所以需要先启动 Hadoop,然后再进行操作。
# 运行 Flume
cd $FLUME_HOME
# 注意引用路径,需要修改成你自己的
./bin/flume-ng agent -n a1 -c conf/ -f job/httpToHDFS.conf -Dflume.root.logger=INFO,console
Flume 启动完成后,如下所示:
我们通过其它窗口,使用 curl
命令向 HTTP 源发送两条模拟数据:
curl -X POST -d '[{"body":"hello body"}]' http://localhost:5140
curl -X POST -d '[{"body":"HELLO FLUME"}]' http://localhost:5140
数据发送完成后,Flume 会采集到该数据,并存储到 HDFS 上。
通过命令,查看 HDFS 中存储的内容,验证拦截器是否生效:
hdfs dfs -text /flume/events/2024-04-04/1630/00/ev*
结果如下所示:
可以看到,我们在上面分别发送了两条数据 hello body
和 HELLO FLUME
,但最终 HDFS 中只存储了一条数据。
这是因为我们设置的拦截器生效了,它对数据中包含 hello
字符串的事件进行了过滤,故只存储了一条数据。
Flume 拦截器就是起到这样的效果,对数据进行处理、转换、删除等操作,是不是很简单呀。(同学,包会的呀)。