1数据采集是实时数仓ODS层的一个重要环节。要求使用Flume从日志数据文件中采集数据并写入到Kafka特定主题中。要求使用flume自定义拦截器实现非法日志数据的清洗包括新建maven工程编写拦截器源码编译打jar包并部署等步骤5分

步骤如下：

1. 新建一个Maven工程，可以使用任何IDE（如Eclipse、Intellij IDEA）创建一个Maven项目。

2. 在pom.xml文件中添加Flume和Kafka的依赖：

<dependencies>
    <dependency>
        <groupId>org.apache.flume</groupId>
        <artifactId>flume-ng-core</artifactId>
        <version>1.9.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>2.8.0</version>
    </dependency>
</dependencies>

3. 创建一个新的Java类，用于实现自定义的拦截器。这个类需要实现Flume的Interceptor接口，重写intercept()方法来实现日志数据的清洗逻辑。例如，以下是一个简单的示例：

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

import java.nio.charset.StandardCharsets;
import java.util.List;

public class IllegalLogInterceptor implements Interceptor {

    @Override
    public void initialize() {
        // 初始化方法，可以在这里进行一些初始化操作
    }

    @Override
    public Event intercept(Event event) {
        // 获取原始日志数据
        String logData = new String(event.getBody(), StandardCharsets.UTF_8);

        // 进行非法日志数据的清洗逻辑
        if (isIllegalLog(logData)) {
            // 如果是非法日志数据，则返回null，表示丢弃该事件
            return null;
        }

        // 返回清洗后的日志数据
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        // 对于批量事件的处理，可以遍历每个事件，逐个进行拦截处理
        for (Event event : events) {
            Event interceptedEvent = intercept(event);
            if (interceptedEvent != null) {
                // 将清洗后的事件添加到结果列表中
                resultList.add(interceptedEvent);
            }
        }
        return resultList;
    }

    @Override
    public void close() {
        // 关闭方法，可以在这里进行资源释放操作
    }

    private boolean isIllegalLog(String logData) {
        // 判断日志数据是否合法的逻辑
        // 返回true表示非法日志数据，返回false表示合法日志数据
    }

    public static class Builder implements Interceptor.Builder {
        @Override
        public Interceptor build() {
            return new IllegalLogInterceptor();
        }

        @Override
        public void configure(Context context) {
            // 配置方法，可以在这里进行一些配置参数的读取和初始化操作
        }
    }
}

4. 编译和打包项目，生成一个可执行的JAR包。可以使用Maven的install命令来编译和打包项目：

mvn install

5. 部署JAR包到Flume的插件目录中。将生成的JAR包拷贝到Flume的插件目录下的interceptors子目录中。例如，Flume的插件目录为/path/to/flume/plugins.d/，则将JAR包拷贝到/path/to/flume/plugins.d/interceptors/目录下。

6. 配置Flume的Flume agent，将自定义拦截器应用到数据采集的流程中。在Flume的配置文件中，添加以下内容：

# Flume agent配置
agent.sources = source1
agent.channels = channel1
agent.sinks = sink1

# 配置source
agent.sources.source1.type = exec
agent.sources.source1.command = tail -F /path/to/logfile.log
agent.sources.source1.channels = channel1

# 配置channel
agent.channels.channel1.type = memory
agent.channels.channel1.capacity = 10000
agent.channels.channel1.transactionCapacity = 1000

# 配置sink
agent.sinks.sink1.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.sink1.kafka.topic = mytopic
agent.sinks.sink1.kafka.bootstrap.servers = localhost:9092
agent.sinks.sink1.kafka.producer.acks = 1
agent.sinks.sink1.kafka.flumeBatchSize = 100
agent.sinks.sink1.channel = channel1

# 配置拦截器
agent.sources.source1.interceptors = interceptor1
agent.sources.source1.interceptors.interceptor1.type = com.example.IllegalLogInterceptor$Builder
agent.sources.source1.interceptors.interceptor1.param1 = value1

其中，/path/to/logfile.log为日志文件的路径，mytopic为Kafka的主题名称，localhost:9092为Kafka的地址。

7. 启动Flume agent，开始数据采集和清洗的流程：

flume-ng agent -n agent -c /path/to/flume/conf -f /path/to/flume/conf/flume.conf -Dflume.root.logger=INFO,console

以上步骤是一个简单的示例，实际使用中需要根据具体需求进行调整和扩展