flume扩展

Flume 内置组件不够用？自己写一个，没你想的那么难

Flume 内置的 Source、Sink、拦截器覆盖了大部分场景，但总有例外。

我遇到过的情况：业务方发的日志格式是自定义的二进制协议，内置的 Source 根本不认识；下游存储是个自研的时序数据库，没有现成的 Sink。这时候只能自己写。

说实话，Flume 的扩展接口设计得还算清晰，继承几个类、实现几个方法就能跑起来。这篇文章把三种扩展——拦截器、Source、Sink——的写法都过一遍，代码贴全，配置也附上，你需要的时候直接抄。

开发前准备：开发环境与依赖

Flume 的自定义组件本质上就是写 Java 类，打包成 JAR 扔到 Flume 的 lib 目录下。

依赖配置

在 pom.xml 中添加 Flume 核心依赖（以 1.9.0 为例）：

<dependency>  
    <groupId>org.apache.flume</groupId>  
    <artifactId>flume-ng-core</artifactId>  
    <version>1.9.0</version>  
    <scope>provided</scope>  <!-- 运行时由 Flume 环境提供 -->  
</dependency>

scope 设成 provided，因为 Flume 运行时已经带了这些类，打包的时候别打进去，避免版本冲突。

核心接口

Flume 扩展的核心是实现官方定义的接口，各组件对应的接口如下：

组件类型	需实现的接口 / 继承的类	核心方法
拦截器	`org.apache.flume.interceptor.Interceptor`	`intercept(Event)` 处理单个事件
Source	继承 `AbstractSource`，实现 `PollableSource`	`process()` 产生并发送事件
Sink	继承 `AbstractSink`，实现 `Configurable`	`process()` 从 Channel 消费事件

每个组件都要配一个 Builder 类，Flume 通过 Builder 来实例化你的组件。这块容易忘，后面代码里会看到。

实战一：自定义拦截器（Interceptor）：给 Event 打个标签

拦截器的用处是：数据从 Source 出来、进 Channel 之前，你插一手，改改 Header、改改 Body，或者直接丢掉某些数据。

下面这个例子实现了一个简单的分类器：根据 Body 内容给 Event 的 Header 里加一个 type 字段，后续可以用 Multiplexing Channel Selector 按类型路由到不同的 Channel。

1.代码实现

通过实现org.apache.flume.interceptor.Interceptor来自定义自己的拦截器

public class MyInterceptor implements Interceptor {
    @Override
    public void initialize() {
// 初始化资源，比如建立连接、加载配置
    }

    /**
     * 单个事件拦截
     * @param event
     * @return
     */
    @Override
    public Event intercept(Event event) {
        // 获取头信息
        Map<String,String> headers = event.getHeaders();

        // 获取数据
        String body = new String(event.getBody());

        // 按 Body 前缀分类  
        if (body.startsWith("number:")) {  
            headers.put("type", "number");  // 数字类型  
        } else if (body.startsWith("log:")) {  
            headers.put("type", "log");      // 日志类型  
        } else {  
            headers.put("type", "other");    // 其他类型  
        }  
        return event;  // 返回处理后的 Event

    }

    /**
     * 批量事件拦截
     * @param list
     * @return
     */
    @Override
    public List<Event> intercept(List<Event> list) {
        for (Event event : events) {  
            intercept(event);  
        }  
        return events; 
    }

    @Override
    public void close() {

    }
    // Builder 是必须的，Flume 通过它来创建拦截器实例
    public static class Builder implements Interceptor.Builder{

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

        @Override
        public void configure(Context context) {
					// 从配置文件读取参数（如无参数可空实现） 
        }
    }

}

2. 打包与部署

代码打包为 JAR（如 flume-custom-interceptor.jar）；
1
mvn clean package

将 JAR 复制到 Flume 安装目录的 lib 文件夹下（确保 Flume 能加载类）
1
cp target/flume-custom-interceptor.jar /usr/local/flume/lib/

3. 配置使用拦截器

在 Flume 配置文件中引用自定义拦截器，并结合 Multiplexing Channel Selector 实现按类型路由：



# 定义组件  
agent.sources = customSource  
agent.channels = numChannel logChannel otherChannel  
agent.sinks = numSink logSink otherSink  

# 配置 Source 并启用拦截器  
agent.sources.customSource.type = seq
#拦截器名称
agent.sources.mySource.interceptors = myInterceptor
# 配置拦截器（注意格式：包名+类名$Builder） 因为 Builder 是拦截器的内部静态类。
agent.sources.mySource.interceptors.myInterceptor.type = com.zhanghe.study.custom_flume.interceptor.MyInterceptor$Builder 

# 配置 Channel 选择器（按 type 头信息路由）  
agent.sources.customSource.selector.type = multiplexing  
# 按 Header 中的 type 字段路由
agent.sources.customSource.selector.header = type    
 # type=number → numChannel 
agent.sources.customSource.selector.mapping.number = numChannel  
 # type=log → logChannel  
agent.sources.customSource.selector.mapping.log = logChannel  
# 默认路由  
agent.sources.customSource.selector.default = otherChannel        

# 配置 Channel（内存通道）  
agent.channels.numChannel.type = memory  
agent.channels.logChannel.type = memory  
agent.channels.otherChannel.type = memory  

# 配置 Sink（输出到控制台日志）  
agent.sinks.numSink.type = logger  
agent.sinks.logSink.type = logger  
agent.sinks.otherSink.type = logger  

# 绑定关系  
agent.sources.customSource.channels = numChannel logChannel otherChannel  
agent.sinks.numSink.channel = numChannel  
agent.sinks.logSink.channel = logChannel  
agent.sinks.otherSink.channel = otherChannel

这个组合很实用：拦截器打标签 + Channel Selector 按标签路由，一条采集链路可以分流到多个下游。

4. 验证效果

启动 Flume 后，序列生成器会产生事件，拦截器会按内容添加 type 头信息，最终不同类型的事件会路由到对应的 Channel 和 Sink，控制台会输出分类后的日志。

实战二：自定义Source：从特殊数据源拉数据

内置的 Source 不认识你的数据源，就得自己写。下面这个 Source 做的事情很简单：每秒生成一条数据，格式是 "自定义前缀: 序号"。

1. 代码实现

自定义的Source需要继承AbstractSource，实现Configurable和PollableSource接口

import org.apache.flume.*;  
import org.apache.flume.conf.Configurable;  
import org.apache.flume.source.AbstractSource;  
import org.apache.flume.source.PollableSource;  
import java.util.concurrent.atomic.AtomicInteger;  

public class MySource extends AbstractSource  
        implements PollableSource, Configurable {  

    private String prefix;  // 自定义前缀（从配置文件读取）  
    private AtomicInteger counter = new AtomicInteger(0);  // 计数器  

    // 从配置文件读取参数  
    @Override  
    public void configure(Context context) {  
        // 读取配置参数，默认值为 "custom"  
        prefix = context.getString("prefix", "custom");  
    }  

    // 核心方法：产生事件并发送到 Channel  
    @Override  
    public Status process() throws EventDeliveryException {  
        Status status = Status.READY;  
        try {  
            // 生成自定义事件内容  
            String data = prefix + ": " + counter.incrementAndGet();  
            Event event = EventBuilder.withBody(data.getBytes());  

            // 将事件发送到 Channel（通过 ChannelProcessor）  
            getChannelProcessor().processEvent(event);  
            Thread.sleep(1000);  // 每秒生成一个事件  
        } catch (Exception e) {  
            status = Status.BACKOFF;  // 失败时返回 BACKOFF ，Flume 会稍后重试 
            if (e instanceof Error) {  
                throw (Error) e;  
            }  
        }  
        return status;  
    }  

    // 失败重试间隔增量（默认 0 即可）  
    @Override  
    public long getBackOffSleepIncrement() {  
        return 0;  
    }  

    // 最大重试间隔（默认 0 即可）  
    @Override  
    public long getMaxBackOffSleepInterval() {  
        return 0;  
    }  
}

2. 配置使用自定义 Source


# 定义组件  
agent.sources = customSource  
agent.channels = memoryChannel  
agent.sinks = loggerSink  

# 配置自定义 Source  
agent.sources.mySource.type = com.zhanghe.study.custom_flume.source.MySource 
# 自定义参数（对应代码中的 prefix） 
agent.sources.customSource.prefix = mydata   

# 配置 Channel 和 Sink（复用之前的配置）  
agent.channels.memoryChannel.type = memory  
agent.sinks.loggerSink.type = logger  

# 绑定关系  
agent.sources.customSource.channels = memoryChannel  
agent.sinks.loggerSink.channel = memoryChannel

有个坑要注意：process() 方法里如果抛出异常，Flume 不会自动重试。你得自己 catch 住，返回 Status.BACKOFF，Flume 才会稍后再次调用 process()。我刚写的时候没注意这个，Source 一报错就直接停了。

实战三：自定义Sink：往特殊目标写数据

自定义 Sink 就是把数据从 Channel 取出来，发到你想要的地方。下面这个 Sink 把数据追加写到本地文件里。

1. 代码实现

自定义的Sink需要继承AbstractSink类,实现Configurable接口

import org.apache.flume.*;  
import org.apache.flume.conf.Configurable;  
import org.apache.flume.sink.AbstractSink;  
import org.slf4j.Logger;  
import org.slf4j.LoggerFactory;  
import java.io.FileWriter;  
import java.io.IOException;  
import java.io.PrintWriter;  

public class MySink extends AbstractSink implements Configurable {  
    private static final Logger logger = LoggerFactory.getLogger(FileSink.class);  
    private String filePath;  // 输出文件路径  
    private PrintWriter writer;  

    // 从配置文件读取参数  
    @Override  
    public void configure(Context context) {  
        filePath = context.getString("filePath");  // 必须配置文件路径  
        if (filePath == null) {  
            throw new IllegalArgumentException("filePath 配置不能为空！");  
        }  
    }  

    // 启动 Sink 时初始化文件写入流  
    @Override  
    public void start() {  
        try {  
            writer = new PrintWriter(new FileWriter(filePath, true));  // 追加模式  
        } catch (IOException e) {  
            logger.error("初始化文件写入流失败", e);  
            throw new FlumeException(e);  
        }  
        super.start();  
    }  

    // 核心方法：从 Channel 读取事件并处理  
    @Override  
    public Status process() throws EventDeliveryException {  
        Status status = Status.READY;  
        Channel channel = getChannel();  
        Transaction txn = channel.getTransaction();  // 开启事务  

        try {  
            txn.begin();  // 事务开始  
            Event event = channel.take();  // 从 Channel 读取事件  

            if (event != null) {  
                // 将事件内容写入文件  
                String data = new String(event.getBody());  
                writer.println(data);  
                writer.flush();  // 立即刷新  
            } else {  
                status = Status.BACKOFF;  // 无事件时返回 BACKOFF  
            }  
            txn.commit();  // 事务提交  
        } catch (Exception e) {  
            txn.rollback();  // 失败时回滚事务  
            status = Status.BACKOFF;  
            if (e instanceof Error) {  
                throw (Error) e;  
            }  
        } finally {  
            txn.close();  // 关闭事务  
        }  
        return status;  
    }  

    // 停止时关闭资源  
    @Override  
    public void stop() {  
        if (writer != null) {  
            writer.close();  
        }  
        super.stop();  
    }  
}

2. 配置使用自定义 Sink

# 定义组件  
agent.sources = seqSource  
agent.channels = memoryChannel  
agent.sinks = fileSink  

# 配置 Source（使用序列生成器）  
agent.sources.seqSource.type = seq  

# 配置自定义 Sink  
agent.sinks.fileSink.type = com.zhanghe.study.custom_flume.sink.MySink  
# 输出文件路径  
agent.sinks.fileSink.filePath = /tmp/flume-custom-sink.log  

# 配置 Channel  
agent.channels.memoryChannel.type = memory  

# 绑定关系  
agent.sources.seqSource.channels = memoryChannel  
agent.sinks.fileSink.channel = memoryChannel