flume 自定义json拦截器

Flume是一个分布式、可靠、可扩展的日志收集、聚合、传输系统。它使用拦截器来处理数据流，可以在数据流经过某个拦截器时对数据进行预处理、过滤、转换等操作。

下面给出一个自定义的Flume拦截器，用于将数据转换为JSON格式。该拦截器可以将文本数据转换为JSON格式，并添加时间戳和其他元数据，方便后续处理和分析。

package com.example.flume.interceptor;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;

public class JSONInterceptor implements Interceptor {

    private Gson gson;

    @Override
    public void initialize() {
        gson = new GsonBuilder().create();
    }

    @Override
    public Event intercept(Event event) {
        // 获取原始数据
        String data = new String(event.getBody());
        // 将数据转换为JSON格式
        Map<String, Object> jsonMap = new HashMap<String, Object>();
        jsonMap.put("timestamp", System.currentTimeMillis());
        jsonMap.put("data", data);
        String json = gson.toJson(jsonMap);
        // 构造新的Event
        Event newEvent = EventBuilder.withBody(json.getBytes(), event.getHeaders());
        return newEvent;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        List<Event> interceptedEvents = new ArrayList<Event>();
        for (Event event : events) {
            interceptedEvents.add(intercept(event));
        }
        return interceptedEvents;
    }

    @Override
    public void close() {
        // do nothing
    }

    public static class Builder implements Interceptor.Builder {

        @Override
        public void configure(Context context) {
            // do nothing
        }

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

}

上述代码中，我们首先在`initialize()`方法中创建了一个Gson对象，用于将数据转换为JSON格式。然后在`intercept()`方法中，我们获取原始数据，将其封装成一个Map对象，并添加时间戳等元数据。接着使用Gson将Map对象转换为JSON格式，并构造一个新的Event对象返回。

最后，我们还需要在Flume的配置文件中添加相关配置，以启用该拦截器：

agent.sources = source1
agent.sinks = sink1
agent.channels = channel1

agent.sources.source1.type = netcat
agent.sources.source1.bind = localhost
agent.sources.source1.port = 44444

agent.sinks.sink1.type = logger

agent.channels.channel1.type = memory

# 添加JSONInterceptor拦截器
agent.sources.source1.interceptors = i1
agent.sources.source1.interceptors.i1.type = com.example.flume.interceptor.JSONInterceptor$Builder

agent.sources.source1.channels = channel1
agent.sinks.sink1.channel = channel1