点一下关注吧！！！非常感谢！！持续更新！！！

Java篇开始了！

目前开始更新 MyBatis，一起深入浅出！

目前已经更新到了：

• Hadoop（已更完）
• HDFS（已更完）
• MapReduce（已更完）
• Hive（已更完）
• Flume（已更完）
• Sqoop（已更完）
• Zookeeper（已更完）
• HBase（已更完）
• Redis （已更完）
• Kafka（已更完）
• Spark（已更完）
• Flink（已更完）
• ClickHouse（已更完）
• Kudu（已更完）
• Druid（已更完）
• Kylin（已更完）
• Elasticsearch（已更完）
• DataX（已更完）
• Tez（已更完）
• 数据挖掘（已更完）
• Prometheus（已更完）
• Grafana（已更完）
• 离线数仓（正在更新…）

章节内容

上节我们完成了如下的内容：

• 自定义拦截器(续接上节)
• 采集启动日志和事件日志

日志数据采集小结

（总结前置，下面是正文）

在 Apache Flume 中，拦截器（Interceptor）是数据流管道的一个关键组件，它允许在事件（Event）进入 Flume Channel 之前对其进行修改或过滤。通过自定义拦截器，你可以实现特定的业务逻辑，如数据过滤、字段添加或修改、格式转换等。
自定义拦截器 是指用户根据需求自行编写 Java 代码来扩展 Flume 的功能，而不是使用默认的拦截器。

• 使用 taildir source监控指定多个目录，可以给不同目录的日志加上不同Header
• 在每个目录上可以使用正则匹配多个文件
• 使用自定义拦截器，主要功能是从JSON串种获取时间戳，加到event的header中
• hdfs sink使用event header中的信息写数据（控制写文件的位置）
• hdfs文件的滚动方式（基于文件大小、基于event数量、基于时间）
• 调节Flume JVM内存的分配

工作原理

• 事件生成（Source）： 数据从外部系统通过 Source 进入 Flume。
• 拦截器（Interceptor）： 在 Source 和 Channel 之间拦截数据，执行预处理、过滤或增强。
• 传输（Channel）： 处理后的事件被传递到 Channel，等待被 Sink 消费。
• 消费（Sink）： 最终将数据写入目标系统（如 HDFS、Kafka 等）。

开发和部署注意事项

• 依赖管理： 开发自定义拦截器需要依赖 Flume 的核心库，如 flume-ng-core 和 flume-ng-sdk。
• 测试： 在本地测试拦截器逻辑，确保其功能正确，性能符合预期。
• 部署： 将 JAR 文件上传至 Flume Agent 的 lib 目录并重启 Flume 服务。
• 性能监控： 自定义拦截器可能会影响 Flume 的性能，尤其是在拦截逻辑复杂的情况下。建议在生产环境中监控资源使用情况。

采集启动日志和事件日志

上节我们完成了Agent 的配置，接来我们继续。

自定义拦截器

编码完成后打包上传到服务器，放置在 $FLUME_HOME/lib

编写代码

package icu.wzk;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public class LogTypeInterceptor implements Interceptor {

    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {
        // 获取Event的body
        String eventBody = new String(event.getBody(), StandardCharsets.UTF_8);
        // 获取Event的Header
        Map<String, String> headerMap = event.getHeaders();
        // 解析body获取JSON串
        String[] bodyArr = eventBody.split("\\s+");
        try {
            String jsonStr = bodyArr[6];
            String timestampStr = "";
            JSONObject jsonObject = JSON.parseObject(jsonStr);
            if (headerMap.getOrDefault("logtype", "").equals("start")) {
                // 取启动时间戳
                jsonObject.getJSONObject("app_active").getString("time");
            } else if (headerMap.getOrDefault("logtype", "").equals("event")) {
                // 取事件日志第一条记录的时间戳
                JSONArray jsonArray = jsonObject.getJSONArray("wzk_event");
                if (jsonArray.size() > 0) {
                    timestampStr = jsonArray.getJSONObject(0).getString("time");
                }
            }
            // 将时间戳转换为 yyyy-MM-dd
            long timestamp = Long.parseLong(timestampStr);
            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");
            Instant instant = Instant.ofEpochMilli(timestamp);
            LocalDateTime localDateTime = LocalDateTime.ofInstant(instant, ZoneId.systemDefault());
            String date = formatter.format(localDateTime);
            // 转换后将字符串放置到Header中
            headerMap.put("logtime", date);
            event.setHeaders(headerMap);
        } catch (Exception e) {
            headerMap.put("logtime", "Unknown");
            event.setHeaders(headerMap);
        }
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        List<Event> lstEvent = new ArrayList<>();
        for (Event event: events){
            Event outEvent = intercept(event);
            if (outEvent != null) {
                lstEvent.add(outEvent);
            }
        }
        return lstEvent;
    }

    @Override
    public void close() {

    }

    public static class Builder implements
            Interceptor.Builder {
        @Override
        public Interceptor build() {
            return new LogTypeInterceptor();
        }
        @Override
        public void configure(Context context) {
        }
    }
    
}

打包项目

mvn clean package

打包结果如下，我们需要将“”上传到服务器中：

启动测试

flume-ng agent --conf-file /opt/wzk/flume-conf/flume-log2hdfs3.conf -name a1 -Dflume.roog.logger=INFO,console

启动的结果如下图所示，如果你缺什么文件夹之类的，自己创建出来：

测试结果

写入log文件：

vim /opt/wzk/logs/start/test.log

写入的内容如下所示：

2020-08-02 18:19:32.959 [main] INFO icu.wzk.ecommerce.AppStart - {"app_active":{"name":"app_active","json":{"entry":"1","action":"0","error_code":"0"},"time":1596342840284},"attr":{"area":"大庆","uid":"2F10092A2","app_v":"1.1.15","event_type":"common","device_id":"1FB872-9A1002","os_type":"2.8","channel":"TB","language":"chinese","brand":"iphone-8"}}

写入的结果如下图所示：

写入log文件：

vim /opt/wzk/logs/event/test.log

写入的内容如下所示：

2020-08-02 18:20:11.877 [main] INFO icu.wzk.ecommerce.AppEvent - {"wzk_event":[{"name":"goods_detail_loading","json":{"entry":"1","goodsid":"0","loading_time":"93","action":"3","staytime":"56","showtype":"2"},"time":1596343881690},{"name":"loading","json":{"loading_time":"15","action":"3","loading_type":"3","type":"1"},"time":1596356988428},{"name":"notification","json":{"action":"1","type":"2"},"time":1596374167278},{"name":"favorites","json":{"course_id":1,"id":0,"userid":0},"time":1596350933962}],"attr":{"area":"长治","uid":"2F10092A4","app_v":"1.1.14","event_type":"common","device_id":"1FB872-9A1004","os_type":"0.5.0","channel":"QL","language":"chinese","brand":"xiaomi-0"}}

写入的结果如下图所示：

查看结果

控制台已经输出了结果：

我们查看HDFS，也输出了对应的内容出来：

生产环境

生产环节中，推荐使用：

nohup flume-ng agent --conf-file /opt/wzk/flume-conf/flume-log2hdfs3.conf -name a1 -Dflume.roog.logger=INFO,console > dev/null 2>&1 &

• nohup 该命令允许用户退出账户、关闭终端之后还继续运行相应的进程
• /dev/null 代表Linux的空设备文件，所有往这个文件里面写入的内容都会丢失，也称黑洞
• 标准输入0，从键盘获得输入 /proc/self/fd/0
• 标准输出1，输出到屏幕（控制台）/proc/self/fd/1
• 错误输出2，输出到屏幕（控制台）/proc/self/fd/2
• /dev/null 标准输出1重定向到 /dev/null 中，此时标准输出不存在，没有任何地方能够找到输出的内容
• 2>&1 错误输出将会和标准输出输出到同一个地方
• /dev/null 2>&1 不会输出任何信息到控制台，也不会有任何信息输出到文件中