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; // 是否启用拦截器

    // 初始化（可读取配置参数）
    @Override
    public void initialize() {
        isEnabled = true; // 可从Context中读取配置，如context.getBoolean("enable.timestamp", true)
    }

    // 单个事件处理：添加时间戳到Headers
    @Override
    public Event intercept(Event event) {
        if (isEnabled) {
            Map<String, String> headers = new HashMap<>(event.getHeaders());
            headers.put("timestamp", System.currentTimeMillis() + "");
            event.setHeaders(headers);
        }
        return event;
    }

    // 批量事件处理（优化批量操作）
    @Override
    public List<Event> intercept(List<Event> events) {
        List<Event> interceptedEvents = new ArrayList<>();
        for (Event event : events) {
            interceptedEvents.add(intercept(event));
        }
        return interceptedEvents;
    }

    // 资源清理（本例无资源需释放）
    @Override
    public void close() {}

    // 拦截器构建器（工厂模式）
    public static class Builder implements Interceptor.Builder {
        @Override
        public Interceptor build() {
            return new TimestampInterceptor();
        }

        @Override
        public 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;
}