大数据-227 离线数仓 - Flume 自定义拦截器(续接上节) 采集启动日志和事件日志

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

Java篇开始了！

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

目前已经更新到了：

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

章节内容

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

  • Flume 自定义拦截器
  • 拦截原理 拦截器实现 Java

在这里插入图片描述

自定义拦截器

（续接上节，上节已经到了打包的部分）

上传结果

将刚才的打包上传到这个目录下：
我拷贝的是带依赖的：“flume-test-1.0-SNAPSHOT-jar-with-dependencies.jar”

/opt/servers/apache-flume-1.9.0-bin/lib/

测试效果

我们创建刚才说的conf文件：

vim /opt/wzk/flume-conf/flumetest1.conf

编写的内容如下图所示：
在这里插入图片描述
启动进行测试：

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

启动结果如下图所示：
z

我们启动 telnet 来传入数据：

telnet h122.wzk.icu 9999

启动之后输入数据：

2020-07-30 14:18:47.339 [main] INFO com.ecommerce.AppStart - {"app_active":{"name":"app_active","json":{"entry":"1","action":"1","error_code":"0"},"time":1596111888529},"attr":{"area":"泰安","uid":"2F10092A9","app_v":"1.1.13","event_type":"common","device_id":"1FB872-9A1009","os_type":"4.7.3","channel":"DK","language":"chinese","brand":"iphone-9"}}

此时控制台的数据内容部分为：

4/08/27 15:31:31 INFO source.NetcatSource: Source starting
24/08/27 15:31:31 INFO source.NetcatSource: Created serverSocket:sun.nio.ch.ServerSocketChannelImpl[/172.16.1.130:9999]
24/08/27 15:32:09 INFO sink.LoggerSink: Event: { headers:{logtime=2020-07-30} body: 32 30 32 30 2D 30 37 2D 33 30 20 31 34 3A 31 38 2020-07-30 14:18 }

对应的截图如下图所示：
在这里插入图片描述

采集启动日志（使用自定义拦截器）

配置文件

新建一个配置文件：

vim /opt flume-log2hdfs2.conf

写入的内容如下所示：

a1.sources = r1
a1.sinks = k1
a1.channels = c1
# taildir source
a1.sources.r1.type = TAILDIR
a1.sources.r1.positionFile = /opt/wzk/conf/startlog_position.json
a1.sources.r1.filegroups = f1
a1.sources.r1.filegroups.f1 = /opt/wzk/logs/start/.*log
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = icu.wzk.CustomerInterceptor$Builder
# memorychannel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 100000
a1.channels.c1.transactionCapacity = 2000
# hdfs sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /user/data/logs/start/dt=%{logtime}/
a1.sinks.k1.hdfs.filePrefix = startlog.
a1.sinks.k1.hdfs.fileType = DataStream
# 配置文件滚动方式（文件大小32M）
a1.sinks.k1.hdfs.rollSize = 33554432
a1.sinks.k1.hdfs.rollCount = 0
a1.sinks.k1.hdfs.rollInterval = 0
a1.sinks.k1.hdfs.idleTimeout = 0
a1.sinks.k1.hdfs.minBlockReplicas = 1
# 向hdfs上刷新的event的个数
a1.sinks.k1.hdfs.batchSize = 1000
# 使用本地时间
# a1.sinks.k1.hdfs.useLocalTimeStamp = true
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

内容的截图如下所示：
在这里插入图片描述
修改的内容如下：

  • 给source增加自定义拦截器
  • 去掉时间戳 a1.sinks.k1.hdfs.useLocalTimeStamp = true
  • 根据header中的logtime写文件

测试运行

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

拷贝日志

修改日志的内容：

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

继续写入如下的内容：

2020-07-30 14:18:47.339 [main] INFO com.ecommerce.AppStart - {"app_active":{"name":"app_active","json":{"entry":"1","action":"1","error_code":"0"},"time":1596111888529},"attr":{"area":"泰安","uid":"2F10092A9","app_v":"1.1.13","event_type":"common","device_id":"1FB872-9A1009","os_type":"4.7.3","channel":"DK","language":"chinese","brand":"iphone-9"}}

写入内容如下图所示：
在这里插入图片描述

测试效果

可以看到HDFS上，已经有了读出日志中的时间的内容：
在这里插入图片描述

采集启动日志和事件日志

本系统中要采集两种日志：

  • 启动日志
  • 事件日志
    不同的日志放置在不同的目录下，要想一次拿到全部日志需要监控多个目录。
    在这里插入图片描述

总体思路

  • taildir 监控多个目录
  • 修改自定义拦截器，不同来源的数据加上不同标志
  • HDFS、Sink 根据标志写文件

Agent 介绍

Flume 是一个分布式、高可靠、可用来收集、聚合和传输大量日志数据的系统。在 Flume 的体系结构中，Agent 是一个关键的组件。每个 Agent 是一个独立的 JVM 进程，负责从数据源获取数据并将其传递到下游（如文件系统、数据库、或者另一个 Agent）

Agent 的核心组成部分

Flume Agent 的架构是高度模块化的，它由以下三个核心组件构成：

Source (源)

Source 是数据流的起点，负责接收外部数据。它支持多种数据传输协议和格式，能够从日志文件、网络端口、消息队列等数据源中接收事件。

支持的 Source 类型：

  • Avro Source：接收来自其他 Flume Agent 的 Avro 格式数据。
  • Syslog Source：处理 Syslog 消息。
  • Exec Source：从本地命令或脚本的输出中读取数据。
  • HTTP Source：通过 HTTP 接口接收数据。
  • Spooling Directory Source：监控特定目录中的文件并读取内容。

Channel (通道)

Channel 是 Agent 的数据缓冲区域，用于在 Source 和 Sink 之间暂存事件。Channel 的设计保证了在数据流动中断（如网络故障）时，数据不会丢失。

常见的 Channel 类型：

  • Memory Channel：将事件存储在内存中，速度快，但可能会丢失数据（如果 Agent 崩溃）。
  • File Channel：将事件存储在磁盘文件中，提供高可靠性但性能较低。
  • Kafka Channel：使用 Kafka 作为中转通道，适合分布式场景。

Sink (接收器)

Sink 是数据流的终点，负责将事件传递到下游存储或处理系统。

支持的 Sink 类型：

  • HDFS Sink：将事件写入 Hadoop HDFS。
  • Kafka Sink：将事件发送到 Kafka。
  • Elasticsearch Sink：将事件写入 Elasticsearch。
  • Logger Sink：将事件输出到日志。
  • Avro Sink：将事件传递到另一个 Flume Agent 的 Avro Source。

Agent配置

vim /opt/wzk/flume-conf/flume-log2hdfs3.conf

写入的内容如下图所示：

a1.sources = r1
a1.sinks = k1
a1.channels = c1
# taildir source
a1.sources.r1.type = TAILDIR
a1.sources.r1.positionFile = /opt/wzk/conf/startlog_position.json
a1.sources.r1.filegroups = f1 f2
a1.sources.r1.filegroups.f1 = /opt/wzk/logs/start/.*log
a1.sources.r1.headers.f1.logtype = start
a1.sources.r1.filegroups.f2 = /opt/wzk/logs/event/.*log
a1.sources.r1.headers.f2.logtype = event
# 自定义拦截器
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = icu.wzk.LogTypeInterceptor$Builder
# memorychannel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 100000
a1.channels.c1.transactionCapacity = 2000
# hdfs sink
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = /user/data/logs/%{logtype}/dt=%
{logtime}/
a1.sinks.k1.hdfs.filePrefix = startlog.
a1.sinks.k1.hdfs.fileType = DataStream
# 配置文件滚动方式（文件大小32M）
a1.sinks.k1.hdfs.rollSize = 33554432
a1.sinks.k1.hdfs.rollCount = 0
a1.sinks.k1.hdfs.rollInterval = 0
a1.sinks.k1.hdfs.idleTimeout = 0
a1.sinks.k1.hdfs.minBlockReplicas = 1
# 向hdfs上刷新的event的个数
a1.sinks.k1.hdfs.batchSize = 1000
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
  • filegroups：指定filegroups，可以有多个，以空格分割（taildir source可以同时监控多个目录中的文件）
  • headers.filegroups.headerKey

给Event增加header Key，不同的filegroup，可配置不同的value。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/diannao/61109.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

windows系统中实现对于appium的依赖搭建

windows系统中实现对于appium的依赖搭建

Node.js：Appium是基于Node.js的，因此需要安装Node.js。可以从Node.js官网下载并安装。 Java Development Kit (JDK)：用于Android应用的自动化测试，需要安装JDK。可以从Oracle官网下载并安装。 Android SDK：进行Andro…
阅读更多...
【langchain4j】AIservices能够实现更加灵活的chain

【langchain4j】AIservices能够实现更加灵活的chain

文章目录 AI service介绍如何工作的AiServices提供的能力支持的返回形式 简单的例子：接收用户消息，并按规定返回接收单个变量接收更多动态变量 advanced RAGChaining multiple AI Services：多个AiSerives合并到一起相关教程：[Lang…
阅读更多...
蓝桥杯某例题的解决方案和拓展（完全能解决例题本身）

蓝桥杯某例题的解决方案和拓展（完全能解决例题本身）

蓝桥杯题目：求1（包含）直到20230408（包含）所有自然数的加和。 这个题比较恶心的一点在于，20230408本身没有超过int的上限，但是它的加和是超过int上限的，因此如果直接用int来计算&…
阅读更多...
Ubuntu24.04安装gpfs客户端

Ubuntu24.04安装gpfs客户端

文章目录 Ubuntu24.04安装gpfs客户端拷贝软件包在客户端执行命令，提取产品包进入安装包目录，安装相关产品包编译。编译过程中会检查系统依赖接入集群（后续） Ubuntu24.04安装gpfs客户端 拷贝软件包 scp /root/Spectrum_Scale_Dat…
阅读更多...
【UGUI】背包的交互01（道具信息跟随鼠标+道具信息面板显示）

【UGUI】背包的交互01（道具信息跟随鼠标+道具信息面板显示）

详细程序逻辑过程 初始化物品栏： 在 Awake 方法中，通过标签找到提示框和信息面板。 循环生成10个背包格子，并为每个格子设置图标和名称。 为每个格子添加 UInterMaager232 脚本，以便处理交互事件。 关闭提示框和信息面板&#…
阅读更多...
同步互斥相关习题10道 附详解

同步互斥相关习题10道 附详解

PV操作 2016 某系统允许最多10个进程同时读文件F，当同时读文件F的进程不满10个时，欲读该文件的其他文件可立即读，当已有10个进程在读文件F时读，其他欲读文件F的进程必须等待，直至有进程读完后退出方可去读 在实现管…
阅读更多...
Postman之数据提取

Postman之数据提取

Postman之数据提取 1. 提取请求头\request中的数据2. 提取响应消息\response中的数据3. 通过正在表达式提取4. 提取cookies数据 本文主要讲解利用pm对象对数据进行提取操作，虽然postman工具的页面上也提供了一部分的例子，但是实际使用时不是很全面&#…
阅读更多...
(蓝桥杯C/C++)——动态规划(DP)

(蓝桥杯C/C++)——动态规划(DP)

目录 一、线性DP 1.DP(动态规划)简介 2.动态规划的分析步骤 3.例题讲解 二、二维DP 1.二维DP简介 2.选数异或 三、最长上升子序列LIS 1.LIS简介 2.例题讲解 四、最长公共子序列LCS 1.最长公共子序列 2.最长公共子序列 2.求出具体子序列 一、线性DP 1.DP(动态规划)…
阅读更多...
【专题】数据库原理与应用之故障恢复

【专题】数据库原理与应用之故障恢复

1. 数据库故障恢复概述 数据库的可恢复性： DBMS能把数据库从被破坏、不正确的状态、恢复到最近一个正确的状态。 恢复管理任务的种类： 一是在未发生故障而系统正常运行时，采取一些必要措施为恢复工作打基础。 二是在发生故障后进行恢复处…
阅读更多...
EXCEL 或 WPS 列下划线转驼峰

EXCEL 或 WPS 列下划线转驼峰

使用场景： 需要将下划线转驼峰，直接在excel或wps中第一行使用公式，然后快速刷整个列格式即可。全列工下划线转为格式，使用效果如下： 操作步骤： 第一步：在需要显示驼峰的一列，复制以…
阅读更多...
MODBUS TCP转CANOpen网关

MODBUS TCP转CANOpen网关

Modbus TCP转CANopen网关 型号：SG-TCP-COE-210 产品用途 本网关可以实现将CANOpen接口设备连接到MODBUS TCP网络中；并且用户不需要了解具体的CANOpen和Modbus TCP 协议即可实现将CANOpen设备挂载到MODBUS TCP接口的 PLC上，并和CANOpen设备…
阅读更多...
如何查看本地的个人SSH密钥

如何查看本地的个人SSH密钥

1.确保你的电脑上安装了 Git。 你可以通过终端或命令提示符输入以下命令来检查： git --version 如果没有安装，请前往 Git 官网 下载并安装适合你操作系统的版本。 2.查找SSH密钥 默认情况下，SSH密钥存储在你的用户目录下的.ssh文件夹中。…
阅读更多...
分布式----Ceph部署

分布式----Ceph部署

目录 一、存储基础 1.1 单机存储设备 1.2 单机存储的问题 1.3 商业存储解决方案 1.4 分布式存储（软件定义的存储 SDS） 1.5 分布式存储的类型 二、Ceph 简介 三、Ceph 优势 四、Ceph 架构 五、Ceph 核心组件 #Pool中数据保存方式支持两种类型&…
阅读更多...
自动驾驶系列—面向自动驾驶的模型迭代：工具、平台与最佳实践

自动驾驶系列—面向自动驾驶的模型迭代：工具、平台与最佳实践

🌟🌟 欢迎来到我的技术小筑，一个专为技术探索者打造的交流空间。在这里，我们不仅分享代码的智慧，还探讨技术的深度与广度。无论您是资深开发者还是技术新手，这里都有一片属于您的天空。让我们在知识的海洋中…
阅读更多...
Spring Boot3自定义starter

Spring Boot3自定义starter

1、加入必要依赖 plugins {id javaid org.springframework.boot version 3.2.6id io.spring.dependency-management version 1.1.5 } group org.example.test.starter version 1.1.0jar{enabledtrue// resolveMainClassName }java {toolchain {languageVersion JavaLanguage…
阅读更多...
AI开发-计算机视觉库-OpenCV

AI开发-计算机视觉库-OpenCV

1 需求 官网：OpenCV - Open Computer Vision Library 2 接口 3 示例 import cv2image cv2.imread("./data/train/1_1.jpg") print(type(image)) 4 参考资料
阅读更多...
LuckySheet协同编辑后端示例(Django+Channel,Websocket通信)

LuckySheet协同编辑后端示例(Django+Channel,Websocket通信)

其他的配置不赘述了，直接网上搜一下django使用channel的配置方法 我初步研究通道先用的内存 在setting文件中 CHANNEL_LAYERS {"default": {"BACKEND": "channels.layers.InMemoryChannelLayer",}, } 这里我重点说一下luckyshe…
阅读更多...
ACE之单例

ACE之单例

单例簇 使用双重锁检查优化 #mermaid-svg-RMOXQ0KMo0VnJe7V {font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}#mermaid-svg-RMOXQ0KMo0VnJe7V .error-icon{fill:#552222;}#mermaid-svg-RMOXQ0KMo0VnJe7V .error-text{fill:#552222…
阅读更多...
delphi fmx android 离线人脸识别

delphi fmx android 离线人脸识别

搜遍全网都没有找到delphi android 能用的 离线人脸识别，无需注册什么开发者 有这方面需求的可以用fsdk 这边用的luxand.FSDK8.0 android下的注册号要自己找下 1，用老猫的工具将android 下的sdk，FSDK.java 编译成FSDK.jar 老猫的工具 2，用上面的工具将FSDK.jar 生成de…
阅读更多...
RabbitMQ教程：工作队列（Work Queues）（二）

RabbitMQ教程：工作队列（Work Queues）（二）

RabbitMQ教程：工作队列（Work Queues）（二） 一、引言 在快节奏的软件开发世界中，我们经常面临需要异步处理任务的场景，比如在Web应用中处理耗时的图片处理或数据分析任务。这些任务如果直接在用…
阅读更多...
最新文章

Copyright @ 2022~2023