@[toc]
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 拦截器就是起到这样的效果，对数据进行处理、转换、删除等操作，是不是很简单呀。（同学，包会的呀）。
