未记录事实核验 · 状态：维护中

记录仅说明核验时间与维护状态，不保证适用于所有环境；使用前请留意正文中的版本、适用条件与风险提示。

warning 适用范围与待维护事项

本文适用于已掌握 Flume Agent、Channel、HDFS 配置及 Java 编译打包的读者，示例不是可直接上线的验证结果。原稿未注明 Flume/Hadoop 版本，也未提供启动、路由及文件读取测试；修正后的参数仍须对照目标版本核验。自定义拦截器还需部署 JAR 并验证类加载；脱敏仅按文本模式匹配，不覆盖 Header、所有敏感字段或合规要求。Nginx 示例仅将 500、503 分入错误通道，内存通道有故障丢失风险，容量与吞吐数字不是性能保证。

warning 关联材料边界

以下原引用目标缺失或无法唯一确认，已取消跳转并保留提及：05-Sink输出组件。相关素材/延伸说明暂不可用，不能据此声称已经提供配套材料。

拦截器与序列化

拦截器的定位

拦截器（Interceptor）是 Flume Source 和 Channel 之间的数据预处理层。每个 Event 从 Source 流出后，先经过拦截器链的处理，再进入 Channel。

拦截器能做三件事：

丰富元数据：向 Event Header 写入时间戳、主机名、自定义标签
过滤数据：根据正则匹配决定是否丢弃某条 Event
提取字段：从 Event Body 中抓取值写入 Header，供 Multiplexing Selector 路由使用

拦截器在 Source 级别配置，一个 Source 可以挂多个拦截器形成拦截器链，按配置顺序依次执行。

拦截器接口规范

所有拦截器都实现 org.apache.flume.interceptor.Interceptor 接口：

public interface Interceptor {
    // 初始化资源（读取配置、编译正则等）
    void initialize();

    // 处理单条 Event，返回 null 表示丢弃此 Event
    Event intercept(Event event);

    // 批量处理，通常循环调用单条版本
    List<Event> intercept(List<Event> events);

    // 释放资源
    void close();

    // 必须实现的内部 Builder，负责从配置文件构造拦截器实例
    interface Builder extends Configurable {
        Interceptor build();
    }
}

Builder 内部类模式是 Flume 拦截器的标准构造方式。Flume 框架通过反射调用 Builder.build() 创建实例，配置项通过 Configurable.configure(Context) 注入。自定义拦截器必须遵循这个模式。

内置拦截器详解

Timestamp 拦截器

为每个 Event 的 Header 中写入当前 Unix 时间戳（毫秒），Key 固定为 timestamp。在未启用 hdfs.useLocalTimeStamp = true 时，HDFS Sink 按时间分区（/data/%Y-%m-%d/%H）需要 Event Header 中有可解析的 timestamp；Timestamp 拦截器是提供该字段的一种方式。启用本地时间选项时，Sink 使用本地时间而非该 Header。

agent.sources.r1.interceptors = ts
agent.sources.r1.interceptors.ts.type = timestamp

# 如果 Event 已有 timestamp Header，是否保留原值（默认 false = 覆盖）
agent.sources.r1.interceptors.ts.preserveExisting = false

preserveExisting = true 适用于多级 Agent 架构：采集端已打时间戳，聚合端不需要重新打，保留原始采集时间。

Host 拦截器

将运行 Flume Agent 的主机名或 IP 写入 Event Header，默认 Key 为 host。在多机聚合场景中，聚合端收到来自多台机器的数据，通过 host Header 可以溯源数据来自哪台采集机。

agent.sources.r1.interceptors = hi
agent.sources.r1.interceptors.hi.type = host

# true = 写入 IP 地址，false = 写入主机名（默认 true）
agent.sources.r1.interceptors.hi.useIP = false

# Header 的 Key 名称（默认 "host"）
agent.sources.r1.interceptors.hi.hostHeader = host

# 如果已存在同名 Header，是否保留（默认 false = 覆盖）
agent.sources.r1.interceptors.hi.preserveExisting = false

Static 拦截器

向所有 Event 写入固定的 KV 对 Header，是打业务标签最简单的方式。适合区分数据来源类型（如 log_type=nginx、env=prod）。

agent.sources.r1.interceptors = si
agent.sources.r1.interceptors.si.type = static

# 要写入的 Header Key
agent.sources.r1.interceptors.si.key = log_type

# 要写入的 Header Value（静态字符串）
agent.sources.r1.interceptors.si.value = nginx_access

# 是否保留已存在的同名 Header（默认 true）
agent.sources.r1.interceptors.si.preserveExisting = true

典型用法：多个 Source 分别采集不同业务日志，每个 Source 用 Static 拦截器打上 app_name 标签，写入同一个 KafkaChannel 后下游按 Header 过滤。

Regex Filtering 拦截器

用正则表达式匹配 Event Body，决定保留还是丢弃。适合在采集端过滤掉噪音数据（如 DEBUG 级别日志、健康检查请求），减少下游存储和计算压力。

agent.sources.r1.interceptors = rf
agent.sources.r1.interceptors.rf.type = REGEX_FILTER

# 正则表达式，匹配 Event Body（UTF-8 字符串）
agent.sources.r1.interceptors.rf.regex = ^(DEBUG|TRACE).*

# excludeEvents = true：匹配的 Event 被丢弃（黑名单模式）
# excludeEvents = false：只保留匹配的 Event（白名单模式）
agent.sources.r1.interceptors.rf.excludeEvents = true

过滤 DEBUG/TRACE 行：regex = ^(DEBUG|TRACE).*，excludeEvents = true

只保留 ERROR 行：regex = .*ERROR.*，excludeEvents = false

warning 过滤在 Channel 写入前执行

被丢弃的 Event 不会进入 Channel，也不会到达 Sink。如果业务上对丢弃的数据有审计需求，不要在 Source 端过滤，应在下游处理层过滤。

Regex Extractor 拦截器

从 Event Body 用正则表达式提取子串，写入 Header。这是实现动态路由的关键：先用 Regex Extractor 从日志体中提取字段，再配合 Multiplexing Selector 按 Header 值路由到不同 Channel。

agent.sources.r1.interceptors = re
agent.sources.r1.interceptors.re.type = REGEX_EXTRACTOR

# 正则，捕获组对应下面的 serializers 列表
agent.sources.r1.interceptors.re.regex = ^(\\S+) .* "(GET|POST|PUT|DELETE) (\\S+).*" (\\d{3})
# properties 文件中用双反斜杠保留正则所需的单反斜杠

# 每个捕获组映射到一个 Header Key
agent.sources.r1.interceptors.re.serializers = s1 s2 s3 s4
agent.sources.r1.interceptors.re.serializers.s1.name = client_ip
agent.sources.r1.interceptors.re.serializers.s2.name = http_method
agent.sources.r1.interceptors.re.serializers.s3.name = request_path
agent.sources.r1.interceptors.re.serializers.s4.name = status_code

对于 Nginx 日志 192.168.1.1 - - [date] "GET /api/v1/users" 200 ...，上述配置会提取：

Header: client_ip    = 192.168.1.1
Header: http_method  = GET
Header: request_path = /api/v1/users
Header: status_code  = 200

自定义拦截器：脱敏拦截器完整示例

内置拦截器不足以满足所有需求时，可以实现 Interceptor 接口编写自定义逻辑。以下是一个对日志中手机号和身份证号进行打码的脱敏拦截器：

package com.example.flume.interceptor;

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;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class DesensitizeInterceptor implements Interceptor {

    // 手机号：1开头的11位数字
    private static final Pattern PHONE_PATTERN =
        Pattern.compile("(?<![\\d])(1[3-9]\\d{9})(?![\\d])");

    // 身份证：15位或18位
    private static final Pattern ID_CARD_PATTERN =
        Pattern.compile("(?<![\\d])(\\d{15}|\\d{17}[0-9Xx])(?![\\d])");

    @Override
    public void initialize() {
        // 无需初始化资源
    }

    @Override
    public Event intercept(Event event) {
        if (event == null) return null;

        String body = new String(event.getBody(), StandardCharsets.UTF_8);

        // 手机号：保留前3位和后4位，中间4位用*替换
        Matcher phoneMatcher = PHONE_PATTERN.matcher(body);
        StringBuffer sb = new StringBuffer();
        while (phoneMatcher.find()) {
            String phone = phoneMatcher.group(1);
            String masked = phone.substring(0, 3) + "****" + phone.substring(7);
            phoneMatcher.appendReplacement(sb, Matcher.quoteReplacement(masked));
        }
        phoneMatcher.appendTail(sb);
        body = sb.toString();

        // 身份证：只保留前6位（地区码），其余替换为*
        Matcher idMatcher = ID_CARD_PATTERN.matcher(body);
        sb = new StringBuffer();
        while (idMatcher.find()) {
            String id = idMatcher.group(1);
            String masked = id.substring(0, 6) + (id.length() == 15 ? "*********" : "************");
            idMatcher.appendReplacement(sb, Matcher.quoteReplacement(masked));
        }
        idMatcher.appendTail(sb);

        event.setBody(sb.toString().getBytes(StandardCharsets.UTF_8));
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        List<Event> result = new ArrayList<>(events.size());
        for (Event event : events) {
            Event processed = intercept(event);
            if (processed != null) {
                result.add(processed);
            }
        }
        return result;
    }

    @Override
    public void close() {
        // 无需释放资源
    }

    // Flume 通过 Builder 构造拦截器实例
    public static class Builder implements Interceptor.Builder {

        @Override
        public Interceptor build() {
            return new DesensitizeInterceptor();
        }

        @Override
        public void configure(Context context) {
            // 如需从配置文件读取参数，在此处读取
        }
    }
}

编译打包后，在配置文件中引用：

agent.sources.r1.interceptors = dedup
agent.sources.r1.interceptors.dedup.type = com.example.flume.interceptor.DesensitizeInterceptor$Builder

note 注意 Builder 的引用方式

配置文件中 type 指向的是 Builder 内部类，用 $ 分隔外部类名和内部类名：OuterClass$Builder。这是 Flume 的约定，写成 OuterClass.Builder 会找不到类。

拦截器链的执行顺序

多个拦截器按配置中的空格分隔顺序依次执行：

# 执行顺序：ts → hi → si → rf（从左到右）
agent.sources.r1.interceptors = ts hi si rf

agent.sources.r1.interceptors.ts.type = timestamp
agent.sources.r1.interceptors.hi.type = host
agent.sources.r1.interceptors.si.type = static
agent.sources.r1.interceptors.si.key = env
agent.sources.r1.interceptors.si.value = prod
agent.sources.r1.interceptors.rf.type = REGEX_FILTER
agent.sources.r1.interceptors.rf.regex = ^DEBUG.*
agent.sources.r1.interceptors.rf.excludeEvents = true

本示例的顺序安排（不是所有链路必须遵循的规则；可独立判断且无需审计保留的过滤条件可前置，以减少后续计算）：

打标签类（Timestamp、Host、Static）放前面，后续拦截器可以基于 Header 做条件判断
过滤类（Regex Filter）放后面，先丰富元数据再决定是否丢弃
提取类（Regex Extractor）放在 Static 之后，避免 Header 被覆盖
自定义重计算类（如脱敏）放最后，作用于最终要进 Channel 的数据

性能注意：拦截器在 Source 的处理线程中同步执行，未被丢弃的 Event 按链顺序继续处理；已被过滤掉的 Event 不再进入后续处理。正则计算较重，大流量场景（10万+ TPS）应评估拦截器的 CPU 开销，必要时将复杂处理移到 Flink/Spark 流处理层。

序列化器

序列化器（Serializer）控制 Event 数据以何种格式落地，这里讨论 HDFS Sink 和文件输出 Sink 的输出序列化；反序列化器（Deserializer）负责读取输入并生成 Event，不属于本节的落地格式配置。

BodyTextEventSerializer（默认）

将 Event Body 以纯文本写入文件，Header 信息被忽略。适合写日志文件，文件类型对应 fileType = DataStream。

agent.sinks.k1.type = hdfs
agent.sinks.k1.hdfs.fileType = DataStream
agent.sinks.k1.hdfs.serializer = TEXT
# TEXT 是默认值，通常省略不写

AvroEventSerializer

将 Event（包含 Header 和 Body）序列化为 Avro 格式写入文件，适合需要保留 Header 元数据的场景。Hive 外表或 Spark 读取时需匹配实际 Avro schema；是否比文本读取更高效，应结合数据规模、压缩方式和查询方式实测。

agent.sinks.k1.hdfs.fileType = DataStream
agent.sinks.k1.hdfs.serializer = avro_event

# 可选：压缩 Avro 文件
agent.sinks.k1.hdfs.serializer.compressionCodec = snappy

序列化器选型表

序列化器	fileType 设置	说明	适用场景
`TEXT`（默认）	`DataStream`	纯 Body 文本，一行一条	日志文件、纯文本数据
`avro_event`	`DataStream`	完整 Event（Header + Body）Avro 格式	需要 Header 元数据、结构化数据
不使用本节的 EventSerializer 配置	`SequenceFile`	由 HDFS Sink 的 SequenceFile 写入器输出；`SequenceFile` 是文件类型，不是序列化器名称	与采用相应读取格式的 MapReduce 程序对接
`TEXT`（文本序列化）	`CompressedStream`	由 HDFS Sink 压缩输出，另设 `hdfs.codeC` 并验证目标环境支持的编码器；`CompressedStream` 是文件类型，不是序列化器名称	日志归档，节约存储

综合配置示例：Nginx 日志采集完整链路

以下示例展示拦截器链的实际组合：Timestamp + Host + Regex Extractor，将 Nginx 访问日志按 HTTP 状态码路由到不同 Channel。

agent.sources = nginx_src
agent.channels = c_normal c_error
agent.sinks = k_hdfs k_alert

agent.sources.nginx_src.type = TAILDIR
agent.sources.nginx_src.channels = c_normal c_error
agent.sources.nginx_src.positionFile = /var/flume/nginx_position.json
agent.sources.nginx_src.filegroups = f1
agent.sources.nginx_src.filegroups.f1 = /var/log/nginx/access.log

# 拦截器链：打时间戳 → 打主机名 → 提取状态码
agent.sources.nginx_src.interceptors = ts hi re
agent.sources.nginx_src.interceptors.ts.type = timestamp
agent.sources.nginx_src.interceptors.hi.type = host
agent.sources.nginx_src.interceptors.re.type = REGEX_EXTRACTOR
agent.sources.nginx_src.interceptors.re.regex = .*" (\\d{3}) .*
agent.sources.nginx_src.interceptors.re.serializers = s1
agent.sources.nginx_src.interceptors.re.serializers.s1.name = status_code

# Multiplexing Selector 按状态码路由
agent.sources.nginx_src.selector.type = multiplexing
agent.sources.nginx_src.selector.header = status_code
agent.sources.nginx_src.selector.mapping.200 = c_normal
agent.sources.nginx_src.selector.mapping.201 = c_normal
agent.sources.nginx_src.selector.mapping.500 = c_error
agent.sources.nginx_src.selector.mapping.503 = c_error
agent.sources.nginx_src.selector.default = c_normal

# 正常请求 Channel（内存，高吞吐）
agent.channels.c_normal.type = memory
agent.channels.c_normal.capacity = 100000
agent.channels.c_normal.transactionCapacity = 1000

# 错误请求 Channel（文件，持久化）
agent.channels.c_error.type = file
agent.channels.c_error.checkpointDir = /data/flume/error-checkpoint
agent.channels.c_error.dataDirs = /data/flume/error-data
agent.channels.c_error.capacity = 500000
agent.channels.c_error.transactionCapacity = 5000

# 正常日志写 HDFS
agent.sinks.k_hdfs.type = hdfs
agent.sinks.k_hdfs.channel = c_normal
agent.sinks.k_hdfs.hdfs.path = /logs/nginx/%Y-%m-%d/%H
agent.sinks.k_hdfs.hdfs.fileType = DataStream
agent.sinks.k_hdfs.hdfs.rollInterval = 3600

# 错误日志写告警 HDFS（单独路径）
agent.sinks.k_alert.type = hdfs
agent.sinks.k_alert.channel = c_error
agent.sinks.k_alert.hdfs.path = /logs/nginx-error/%Y-%m-%d
agent.sinks.k_alert.hdfs.fileType = DataStream
agent.sinks.k_alert.hdfs.rollInterval = 600

掌握检查

[ ] 能说出 Interceptor 接口的四个方法及各自职责
[ ] 理解 Builder 内部类为何是必须的，以及配置中 type 值的写法（OuterClass$Builder）
[ ] 能区分 Regex Filtering（丢弃/保留 Event）和 Regex Extractor（提取到 Header）的使用场景
[ ] 能写出拦截器链的配置语法，并解释为何顺序重要
[ ] 理解 TEXT 序列化器和 avro_event 序列化器的区别

练习题

你需要在 Flume 采集端过滤掉所有包含 healthcheck 字符串的 HTTP 请求日志。应该用哪个拦截器，excludeEvents 设为 true 还是 false？写出配置。
系统中有多台机器运行 Flume Agent，数据汇聚到同一个 HDFS 目录后，如何通过 Flume 配置保留每条日志的来源机器信息？
实现一个自定义拦截器，对 Event Body 中匹配 password=\S+ 的部分替换为 password=***。需要实现哪些方法，Builder 类如何编写？

相关文档：Channel 通道机制 | 多级 Agent 架构 | Sink 输出组件（原引用“05-Sink输出组件”未能确定有效目标，暂不可跳转）