Flume 自定义拦截器开发实战：添加时间戳与 JSON 处理

一、引言

在大数据采集领域，Flume 作为分布式日志采集系统，常需对原始数据进行预处理（如清洗、转换、添加元数据等）。

二、开发环境准备

1. 创建 Maven 项目

mvn archetype:generate -DgroupId=com.flume.interceptor -DartifactId=flume-custom-interceptor -DarchetypeArtifactId=maven-archetype-quickstart -DinteractiveMode=false

2. 添加依赖（pom.xml）

<dependencies><!-- Flume核心依赖 --><dependency><groupId>org.apache.flume</groupId><artifactId>flume-ng-core</artifactId><version>1.9.0</version></dependency><!-- JSON解析工具（以fastjson为例） --><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>2.0.28</version></dependency>
</dependencies><!-- 打包插件（阴影打包，包含所有依赖） -->
<build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-shade-plugin</artifactId><version>3.2.4</version><executions><execution><phase>package</phase><goals><goal>shade</goal></goals></execution></executions></plugin></plugins>
</build>

三、自定义时间戳拦截器开发

1. 拦截器代码实现（TimestampInterceptor.java）

package com.flume.interceptor;import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;public class TimestampInterceptor implements Interceptor {private boolean isEnabled; // 是否启用拦截器// 初始化（可读取配置参数）@Overridepublic void initialize() {isEnabled = true; // 可从Context中读取配置，如context.getBoolean("enable.timestamp", true)}// 单个事件处理：添加时间戳到Headers@Overridepublic Event intercept(Event event) {if (isEnabled) {Map<String, String> headers = new HashMap<>(event.getHeaders());headers.put("timestamp", System.currentTimeMillis() + "");event.setHeaders(headers);}return event;}// 批量事件处理（优化批量操作）@Overridepublic List<Event> intercept(List<Event> events) {List<Event> interceptedEvents = new ArrayList<>();for (Event event : events) {interceptedEvents.add(intercept(event));}return interceptedEvents;}// 资源清理（本例无资源需释放）@Overridepublic void close() {}// 拦截器构建器（工厂模式）public static class Builder implements Interceptor.Builder {@Overridepublic Interceptor build() {return new TimestampInterceptor();}@Overridepublic void configure(Context context) {} // 可读取配置文件参数}
}

四、打包与部署

1. 打包命令

mvn clean package -DskipTests

打包后生成flume-custom-interceptor-1.0-SNAPSHOT.jar（含所有依赖）。

2. 部署到 Flume

将打好的 jar 包复制到 Flume 的lib目录：

cp target/flume-custom-interceptor-1.0-SNAPSHOT.jar /opt/flume/lib/

五、模拟数据与 Flume 配置

1. 模拟 JSON 日志生成脚本（createJson.sh）

#!/bin/bash
log='{"host":"www.baidu.com","user_id":"13755569427","items":[{"item_type":"eat","active_time":156234},{"item_type":"car","active_time":156233}]
}'
echo $log >> /home/b.log# 赋予执行权限
chmod +x createJson.sh

2. Flume 配置文件（testInter.conf）

# 定义组件名称
a1.sources = r1
a1.sinks = k1
a1.channels = c1# 配置Source（监控文件）
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /home/b.log
a1.sources.r1.interceptors = i1  # 引用拦截器
a1.sources.r1.interceptors.i1.type = com.flume.interceptor.TimestampInterceptor$Builder  # 完整类名# 配置Channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100# 配置Sink（输出到控制台）
a1.sinks.k1.type = logger# 绑定组件
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

六、运行与验证

1. 启动 Flume 代理

flume-ng agent --name a1 --conf /opt/flume/conf --conf-file /opt/flume/conf/testInter.conf -Dflume.root.logger=INFO,console

2. 生成模拟数据

./createJson.sh  # 多次执行生成多条日志

3. 验证结果

在 Flume 控制台输出中，查看事件 Headers 是否包含timestamp字段：

{"headers": {"timestamp": "16854321000000"  // 时间戳（毫秒级）},"body": "{\"host\":\"www.baidu.com\",\"user_id\":\"13755569427\",...}"
}

七、常见问题与解决

1. 依赖缺失问题

现象：启动 Flume 时报ClassNotFoundException: com.alibaba.fastjson.JSON
原因：未将 fastjson 依赖打入自定义 jar 包。
解决：

使用maven-shade-plugin进行阴影打包（见前文pom.xml配置）。

手动将fastjson.jar复制到 Flume 的lib目录。

2. 拦截器未生效

检查点：

确认 Flume 配置文件中拦截器类名路径正确（包含包名 + 内部类$Builder）。

检查isEnabled开关是否正确读取配置（如从Context获取参数）。

八、扩展场景：JSON 数据解析与校验

若需在拦截器中解析 JSON 并校验数据格式，可结合 fastjson 实现：

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;public Event intercept(Event event) {if (isEnabled) {String json = new String(event.getBody(), StandardCharsets.UTF_8);try {JSONObject obj = JSON.parseObject(json);// 校验必填字段if (!obj.containsKey("user_id")) {throw new IllegalArgumentException("Missing user_id");}// 其他数据清洗逻辑...} catch (Exception e) {// 处理非法数据（如丢弃或标记）event.setBody("INVALID_DATA".getBytes());}// 添加时间戳Map<String, String> headers = new HashMap<>(event.getHeaders());headers.put("timestamp", System.currentTimeMillis() + "");event.setHeaders(headers);}return event;
}