大数据-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，一经查实，立即删除！

相关文章

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

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

文章目录 AI service介绍如何工作的AiServices提供的能力支持的返回形式 简单的例子：接收用户消息，并按规定返回接收单个变量接收更多动态变量 advanced RAGChaining multiple AI Services：多个AiSerives合并到一起相关教程：[Lang…
阅读更多...
【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工具的页面上也提供了一部分的例子，但是实际使用时不是很全面&#…
阅读更多...
【专题】数据库原理与应用之故障恢复

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

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设备…
阅读更多...
分布式----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 参考资料
阅读更多...
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应用中处理耗时的图片处理或数据分析任务。这些任务如果直接在用…
阅读更多...
乐维网管平台（七）：网络稳定与高效的“安全锦囊”

乐维网管平台（七）：网络稳定与高效的“安全锦囊”

试想一下，你给电脑升级了一个软件，升级完成后发现有BUG，经常无故卡死，这时候想回退或重新安装旧版本…相对地，一家企业的网络管理员，在对公司的核心交换机进行复杂的配置调整时，一个小小的疏忽&…
阅读更多...
时代变迁对传统机器人等方向课程的巨大撕裂

时代变迁对传统机器人等方向课程的巨大撕裂

2020年之后，全面转型新质课程规划，传统课程规划全部转为经验。 农耕-代表性生产关系-封建分配制度主要生产力-人力工业-代表性生产关系-资本分配制度工业分为机械时代，电气时代，信息时代；主要生产力-人力转为人脑&…
阅读更多...
Spring6 AOP 面向切面编程

Spring6 AOP 面向切面编程

1. 概念 面向切面编程：一种编程思想。proxy动态代理（实现了这种思想）：在原方法执行时，给原方法的前面或着后面增加其他的方法。增加的方法并不会写在原方法中 原方法就是目标方法，增加的方法就是代理方法 …
阅读更多...
计算机组成与原理（2） basic of computer architecture

计算机组成与原理（2） basic of computer architecture

Instruction Set Architecture (ISA) 和 Hardware System Architecture (HSA) 是计算机体系结构中两个重要的层次，它们各自的职责和作用如下： Instruction Set Architecture (ISA) 定义 ISA是指令集体系结构，是硬件和软件之间的接口。它定义…
阅读更多...
window的wsl（Ubuntu）安装kafka步骤

window的wsl（Ubuntu）安装kafka步骤

环境：Win11 WSL(Linux子系统Ubuntu) apache-zookeeper-3.9.3-bin kafka_2.12-3.8.1 思路：apache上分别下载zookeeper和kafka，在wsl环境安装。在kafka上创建消息的topic，发送消息，接受消息，验证是否安…
阅读更多...
数据结构树和二叉树知识点和递归序列

数据结构树和二叉树知识点和递归序列

二叉树知识点 一.树的概念1.1关于树的名词解释 二.二叉树的概念1. 二叉树性质： 三.满二叉树与完全二叉树递归前序遍历递归中序遍历递归后续遍历 一.树的概念 树是一种非线性数据结构，它是由n个或大于n个的结点来组成具有层次关系的一个集合（…
阅读更多...
速通前端篇 —— CSS

速通前端篇 —— CSS

找往期文章包括但不限于本期文章中不懂的知识点： 个人主页：我要学编程程(ಥ_ಥ)-CSDN博客 所属专栏：速通前端 目录 CSS的介绍 基本语法规范 CSS选择器 标签选择器 class选择器 id选择器 复合选择器 通配符选择器 CSS常见样式 颜…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023