4日志为json数据编写flume拦截器过滤日志不是json的数据

可以使用以下代码编写一个自定义的Flume拦截器，用于过滤掉不是JSON格式的日志数据：

package com.example.flume.interceptor;

import com.google.gson.JsonSyntaxException;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import org.apache.flume.interceptor.InterceptorBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class JsonInterceptor implements Interceptor {

    private static final Logger logger = LoggerFactory.getLogger(JsonInterceptor.class);

    @Override
    public void initialize() {
        // Do nothing
    }

    @Override
    public Event intercept(Event event) {
        try {
            String eventBody = new String(event.getBody(), "UTF-8");
            if (isJson(eventBody)) {
                return event;
            } else {
                logger.warn("Event is not JSON: {}", eventBody);
                return null;
            }
        } catch (Exception e) {
            logger.error("Error intercepting event: {}", e.getMessage());
            return null;
        }
    }

    private boolean isJson(String eventBody) {
        try {
            Gson gson = new Gson();
            gson.fromJson(eventBody, Object.class);
            return true;
        } catch (JsonSyntaxException ex) {
            return false;
        }
    }

    @Override
    public java.util.List<Event> intercept(java.util.List<Event> list) {
        java.util.List<Event> interceptedEvents = new java.util.ArrayList<Event>();
        for (Event event : list) {
            Event interceptedEvent = intercept(event);
            if (interceptedEvent != null) {
                interceptedEvents.add(interceptedEvent);
            }
        }
        return interceptedEvents;
    }

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

    public static class Builder implements InterceptorBuilder {

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

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

然后在Flume配置文件中指定使用这个拦截器，例如：

# Define the source, channel, and sink
agent.sources = mySource
agent.channels = myChannel
agent.sinks = mySink

# Configure the source
agent.sources.mySource.type = exec
agent.sources.mySource.command = tail -F /var/log/myapp.log

# Configure the channel
agent.channels.myChannel.type = memory
agent.channels.myChannel.capacity = 1000
agent.channels.myChannel.transactionCapacity = 100

# Configure the sink
agent.sinks.mySink.type = hdfs
agent.sinks.mySink.hdfs.path = /data/logs/%Y-%m-%d/
agent.sinks.mySink.hdfs.filePrefix = events-
agent.sinks.mySink.hdfs.fileSuffix = .log
agent.sinks.mySink.hdfs.rollInterval = 0
agent.sinks.mySink.hdfs.rollSize = 0
agent.sinks.mySink.hdfs.rollCount = 10000
agent.sinks.mySink.hdfs.writeFormat = Text
agent.sinks.mySink.hdfs.batchSize = 1000
agent.sinks.mySink.hdfs.useLocalTimeStamp = true
agent.sinks.mySink.hdfs.fileType = DataStream
agent.sinks.mySink.channel = myChannel

# Configure the interceptor
agent.sources.mySource.interceptors = myInterceptor
agent.sources.mySource.interceptors.myInterceptor.type = com.example.flume.interceptor.JsonInterceptor$Builder

这个配置将会使用我们编写的JsonInterceptor拦截器，来过滤掉不是JSON格式的日志数据