写一个Java自定义flume拦截器拦截指定文件中的地名数据 还有flume的配置文件

好的，下面是一个Java自定义flume拦截器的示例代码，可以用于拦截指定文件中的地名数据：

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import java.nio.charset.Charset;
import java.util.List;

public class LocationInterceptor implements Interceptor {

    @Override
    public void initialize() {
    }

    @Override
    public Event intercept(Event event) {
        String body = new String(event.getBody(), Charset.forName("UTF-8"));
        if (body.contains("北京") || body.contains("上海") || body.contains("广州")) {
            return event;
        } else {
            return null;
        }
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        for (int i = 0; i < events.size(); i++) {
            Event interceptedEvent = intercept(events.get(i));
            if (interceptedEvent == null) {
                events.remove(i);
                i--;
            } else {
                events.set(i, interceptedEvent);
            }
        }
        return events;
    }

    @Override
    public void close() {
    }

    public static class Builder implements Interceptor.Builder {

        @Override
        public void configure(Context context) {
        }

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

    }

}

上面的代码中，我们定义了一个`LocationInterceptor`类，用于拦截包含指定地名数据的日志事件。在`intercept`方法中，我们获取到事件的内容，判断其中是否包含北京、上海或广州这三个地名，如果包含则返回该事件，否则返回null。在`intercept(List<Event> events)`方法中，我们对事件列表中的每个事件都进行拦截操作，并返回拦截后的事件列表。

接下来，是flume的配置文件示例：

# flume 的配置文件
# 定义 source、channel 和 sink
a1.sources = r1
a1.channels = c1
a1.sinks = k1

# 定义 source 的类型、参数和拦截器
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /path/to/your/log/file
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.example.LocationInterceptor$Builder

# 定义 channel 的类型和参数
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# 定义 sink 的类型、参数和 channel
a1.sinks.k1.type = logger
a1.sinks.k1.channel = c1

# 绑定 source、channel 和 sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

上面的配置文件中，我们定义了一个名为`LocationInterceptor`的拦截器，并将其绑定到source上。在source中，我们使用`exec`类型的source，指定要收集的日志文件路径。在sink中，我们使用`logger`类型的sink，将日志信息输出到控制台。

注意：在使用这个示例代码时，需要将`/path/to/your/log/file`替换成你要收集的日志文件路径。同时，记得将`LocationInterceptor`类所在的包名修改为你自己的包名。