1

我有一个 Flume 组件正在监听 Syslog 流。我做了一个自定义拦截器来修改调用，但它不起作用。我做错了什么？谢谢你，安德里亚

Interceptor 是一个编译良好的 JAR 文件，位于@FLUME_HOME/bin目录中

拦截器类：

package com.test.flume;

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

import java.util.Iterator;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

public class SQLFlumeInterceptor implements Interceptor {

    private final String headerKey;

    private SQLFlumeInterceptor(Context ctx) {            
    }

    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {            
        addPreposition(event);
        return event;
    }

    private void addPreposition(Event event) {            
        System.out.println("Event processed");
        event.setBody( "Modified Event".getBytes() );
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        for (Iterator<Event> iterator = events.iterator(); iterator.hasNext(); ) {

            Event next = iterator.next();
            intercept(next);

            if(next == null) {
            iterator.remove();
            }
        }
        return events;
    }

    @Override
    public void close() {

    }

    public static class CounterInterceptorBuilder implements Interceptor.Builder {

        private Context ctx;

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

        @Override
        public void configure(Context context) {
        this.ctx = context;
        }

  }

水槽.config 文件

# Name the components on this agent
a1.sources = r1
a1.sinks = file-sink
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = syslogtcp
a1.sources.r1.port = 41414
a1.sources.r1.host = 192.168.1.2
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.test.flume.SQLFlumeInterceptor$CounterInterceptorBuilder

# Describe the FILE_ROLLsink
a1.sinks.file-sink.type = FILE_ROLL
a1.sinks.file-sink.sink.directory = /opt/apache-flume-1.5.2-bin/logs/pluto.log
a1.sinks.file-sink.sink.rollInterval = 0
ai.sinks.file-sink.batchSize = 100
ai.sinks.file-sink.fileHeader = true


# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.file-sink.channel = c1

系统将事件记录在文件中而不修改它们，这是相关的调试日志：

2015-04-27 21:39:17,625 (conf-file-poller-0) [DEBUG - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.isValid(FlumeConfiguration.java:313)] Starting validation of configuration for agent: a1, initial-configuration: AgentConfiguration[a1]
SOURCES: {r1={ parameters:{port=41414, host=192.168.1.2, interceptors=i1, interceptors.i1.type=com.test.flume.
SQLFlumeInterceptor$CounterInterceptorBuilder, channels=c1, type=syslogtcp} }}
CHANNELS: {c1={ parameters:{transactionCapacity=100, capacity=1000, type=memory} }}
SINKS: {file-sink={ parameters:{sink.rollInterval=0, type=FILE_ROLL, channel=c1, sink.directory=/opt/apache-flume-1.5.2-bin/logs/pluto.log} }}
4

1 回答 1

2

请将您的拦截器 JAR 文件放在 @FLUME_HOME/lib 目录中，而不是放在 @FLUME_HOME/bin 中。

否则，flume 将不会加载 JAR。

于 2015-04-28T09:15:52.233 回答