Flume 实时数仓 ODS 数据采集与非法日志清洗 - 自定义拦截器实现

步骤如下：\n\n1. 新建 Maven 工程：\n 首先，创建一个新的 Maven 工程。可以使用 IDE（如 Eclipse 或 IntelliJ）创建一个新的 Maven 工程，或者通过命令行执行`mvn archetype:generate`命令创建。\n\n2. 编写拦截器源码：\n 在 Maven 工程中创建一个 Java 类，实现 Flume 的`Interceptor`接口。该接口包含`initialize`、`intercept`和`close`方法。在`intercept`方法中，可以编写逻辑来检查和清洗非法日志数据。例如，可以使用正则表达式或其他方式检查日志数据是否符合特定格式，并将非法数据过滤掉。\n\n java\n import org.apache.flume.Context;\n import org.apache.flume.Event;\n import org.apache.flume.interceptor.Interceptor;\n\n import java.nio.charset.StandardCharsets;\n import java.util.List;\n\n public class IllegalLogInterceptor implements Interceptor {\n\n @Override\n public void initialize() {\n // 初始化操作\n }\n\n @Override\n public Event intercept(Event event) {\n byte[] body = event.getBody();\n String logData = new String(body, StandardCharsets.UTF_8);\n\n // 清洗非法日志数据的逻辑\n if (isIllegalLog(logData)) {\n // 非法日志数据，可以忽略或做其他处理\n return null;\n }\n\n return event;\n }\n\n @Override\n public List<Event> intercept(List<Event> list) {\n for (Event event : list) {\n Event interceptedEvent = intercept(event);\n if (interceptedEvent != null) {\n // 将合法的日志数据添加到结果列表中\n resultList.add(interceptedEvent);\n }\n }\n return resultList;\n }\n\n @Override\n public void close() {\n // 清理资源的操作\n }\n\n private boolean isIllegalLog(String logData) {\n // 非法日志数据的检查逻辑\n // 返回true表示是非法日志数据，返回false表示合法日志数据\n }\n }\n \n\n3. 编译打Jar包：\n 在 Maven 工程的根目录下执行`mvn package`命令，将会编译并打包生成一个 Jar 文件。\n\n4. 部署拦截器：\n 将生成的 Jar 文件拷贝到 Flume 的安装目录的 lib 文件夹下，然后在 Flume 的配置文件中配置使用该拦截器。\n\n properties\n agent.sources = source1\n agent.channels = channel1\n agent.sinks = sink1\n\n agent.sources.source1.type = spooldir\n agent.sources.source1.spoolDir = /path/to/logs\n\n agent.sources.source1.interceptors = interceptor1\n agent.sources.source1.interceptors.interceptor1.type = com.example.IllegalLogInterceptor\n\n agent.channels.channel1.type = memory\n agent.channels.channel1.capacity = 1000\n\n agent.sinks.sink1.type = org.apache.flume.sink.kafka.KafkaSink\n agent.sinks.sink1.kafka.bootstrap.servers = localhost:9092\n agent.sinks.sink1.kafka.topic = my_topic\n\n agent.sources.source1.channels = channel1\n agent.sinks.sink1.channel = channel1\n \n\n 在上述配置中，`source1`表示 Flume 的数据源为`spooldir`（从指定目录的日志文件中读取数据），`interceptor1`表示使用自定义的拦截器。`channel1`是一个内存通道，`sink1`是一个 Kafka 的 sink（将数据写入 Kafka）。\n\n5. 启动 Flume：\n 执行命令`bin/flume-ng agent --conf conf --conf-file conf/flume.conf --name agent -Dflume.root.logger=INFO,console`来启动 Flume。\n\n通过以上步骤，就可以使用 Flume 从日志数据文件中采集数据，并使用自定义拦截器实现非法日志数据的清洗，然后将合法的日志数据写入到指定的 Kafka 主题中。