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;
}