当前位置: 首页 > news >正文

Flume拦截器的实现

Flume conf文件编写

vim file_to_kafka.conf
#定义组件
a1.sources = r1
a1.channels = c1#配置source
a1.sources.r1.type = TAILDIR
a1.sources.r1.filegroups = f1
a1.sources.r1.filegroups.f1 = /Users/zhangjin/model/project/realtime-flink/applog/log/app.*
# 设置断点续传的位置
a1.sources.r1.positionFile = /Users/zhangjin/model/flume/taildir_position.json
a1.sources.r1.interceptors =  i1
a1.sources.r1.interceptors.i1.type = com.flume.interceptor.ETLInterceptor$Builder#配置channel
a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel
a1.channels.c1.kafka.bootstrap.servers = localhost:9092
a1.channels.c1.kafka.topic = topic_log
# 设置不以Flume event 写入数据,以Body数据进行写入
a1.channels.c1.parseAsFlumeEvent = false#组装
a1.sources.r1.channels = c1

Flume ETLInterceptor拦截器的编写

maven依赖

	<dependencies><!--Flume依赖 --><dependency><groupId>org.apache.flume</groupId><artifactId>flume-ng-core</artifactId><version>1.9.0</version><scope>provided</scope></dependency><!--Json格式校验--><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.62</version></dependency></dependencies>

maven package打包依赖

    <build><plugins><plugin><artifactId>maven-compiler-plugin</artifactId><version>2.3.2</version><configuration><source>1.8</source><target>1.8</target></configuration></plugin><plugin><artifactId>maven-assembly-plugin</artifactId><configuration><descriptorRefs><descriptorRef>jar-with-dependencies</descriptorRef></descriptorRefs></configuration><executions><execution><id>make-assembly</id><phase>package</phase><goals><goal>single</goal></goals></execution></executions></plugin></plugins></build>

判断是否是JSON字符串

public class JSONUtil {/** 通过异常判断是否是json字符串* 是:返回true  不是:返回false* */public static boolean isJSONValidate(String log){try {JSONObject.parseObject(log);return true;}catch (JSONException e){return false;}}
}

拦截器实现

  1. 继承Interceptor接口
  2. 实现单event处理
  3. 实现批量event处理
public class ETLInterceptor implements Interceptor {@Overridepublic void initialize() {}/*** 单个event处理* 检验是否是Json格式* @param event* @return*/@Overridepublic Event intercept(Event event) {//1 获取json数据byte[] body = event.getBody();String log = new String(body, StandardCharsets.UTF_8);//2 校验json数据if (JSONUtil.isJSONValidate(log)) {return event;} else {return null;}}/*** 多个event处理* @param list* @return*/@Overridepublic List<Event> intercept(List<Event> list) {Iterator<Event> iterator = list.iterator();while (iterator.hasNext()) {Event event = iterator.next();if (intercept(event) == null) {iterator.remove();}}return list;}@Overridepublic void close() {}/*** 拦截器重写Builder方法*/public static class Builder implements Interceptor.Builder {@Overridepublic Interceptor build() {return new ETLInterceptor();}@Overridepublic void configure(Context context) {}}
}

测试

maven package打包,将生成的jar包放在了Flume的lib目录下
启动kafka

# 启动命令
./bin/kafka-server-start.sh -daemon ./config/server.properties &# 开启消费者
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic_log

启动Flume

bin/flume-ng agent -n a1 -c conf/ -f job/file_to_kafka.conf -Dflume.root.logger=info,console

相关文章:

Flume拦截器的实现

Flume conf文件编写 vim file_to_kafka.conf#定义组件 a1.sources r1 a1.channels c1#配置source a1.sources.r1.type TAILDIR a1.sources.r1.filegroups f1 a1.sources.r1.filegroups.f1 /Users/zhangjin/model/project/realtime-flink/applog/log/app.* # 设置断点续传…...

Swift Combine 学习(四):操作符 Operator

Swift Combine 学习&#xff08;一&#xff09;&#xff1a;Combine 初印象Swift Combine 学习&#xff08;二&#xff09;&#xff1a;发布者 PublisherSwift Combine 学习&#xff08;三&#xff09;&#xff1a;Subscription和 SubscriberSwift Combine 学习&#xff08;四&…...

leetcode 173.二叉搜索树迭代器栈绝妙思路

以上算法题中一个比较好的实现思路就是利用栈来进行实现&#xff0c;以下方法三就是利用栈来进行实现的&#xff0c;思路很好&#xff0c;很简练。进行next的时候&#xff0c;先是一直拿到左边的子树&#xff0c;直到null为止&#xff0c;这一步比较好思考一点&#xff0c;下一…...

df.groupby([pd.Grouper(freq=‘1M‘, key=‘Date‘), ‘Buyer‘]).sum()

df.groupby([pd.Grouper(freq1M, keyDate), Buyer]).sum() 用于根据特定的时间频率和买家&#xff08;Buyer&#xff09;对 DataFrame 进行分组&#xff0c;然后计算每个分组的总和。下面是对这行代码的逐步解释&#xff1a; df.groupby([...])&#xff1a;这个操作会根据传入的…...

LLM - 使用 LLaMA-Factory 部署大模型 HTTP 多模态服务 (4)

欢迎关注我的CSDN&#xff1a;https://spike.blog.csdn.net/ 本文地址&#xff1a;https://spike.blog.csdn.net/article/details/144881432 大模型的 HTTP 服务&#xff0c;通过网络接口&#xff0c;提供 AI 模型功能的服务&#xff0c;允许通过发送 HTTP 请求&#xff0c;交互…...

icp备案网站个人备案与企业备案的区别

个人备案和企业备案是在进行ICP备案时需要考虑的两种不同情况。个人备案是指个人拥有的网站进行备案&#xff0c;而企业备案则是指企业或组织名下的网站进行备案。这两者在备案过程中有一些明显的区别。 首先&#xff0c;个人备案相对来说流程较为简单。个人备案只需要提供个人…...

如何不修改模型参数来强化大语言模型 (LLM) 能力?

前言 如果你对这篇文章感兴趣&#xff0c;可以点击「【访客必读 - 指引页】一文囊括主页内所有高质量博客」&#xff0c;查看完整博客分类与对应链接。 大语言模型 (Large Language Model, LLM, e.g. ChatGPT) 的参数量少则几十亿&#xff0c;多则上千亿&#xff0c;对其的训…...

AF3 AtomAttentionEncoder类的init_pair_repr方法解读

AlphaFold3 的 AtomAttentionEncoder 类中,init_pair_repr 方法方法负责为原子之间的关系计算成对表示(pair representation),这是原子转变器(atom transformer)模型的关键组成部分,直接影响对蛋白质/分子相互作用的建模。 init_pair_repr源代码: def init_pair_repr(…...

DDoS攻击防御方案大全

1. 引言 随着互联网的迅猛发展&#xff0c;DDoS&#xff08;分布式拒绝服务&#xff09;攻击成为了网络安全领域中最常见且危害严重的攻击方式之一。DDoS攻击通过向目标网络或服务发送大量流量&#xff0c;导致服务器过载&#xff0c;最终使其无法响应合法用户的请求。本文将深…...

Vue中常用指令

一、内容渲染指令 1.v-text&#xff1a;操作纯文本&#xff0c;用于更新标签包含的文本&#xff0c;但是使用不灵活&#xff0c;无法拼接字符串&#xff0c;会覆盖文本&#xff0c;可以简写为{{}}&#xff0c;{{}}支持逻辑运算。 用法示例&#xff1a; //把name对应的值渲染到…...

Servlet解析

概念 Servlet是运行在服务端的小程序&#xff08;Server Applet)&#xff0c;可以处理客户端的请求并返回响应&#xff0c;主要用于构建动态的Web应用&#xff0c;是SpringMVC的基础。 生命周期 加载和初始化 默认在客户端第一次请求加载到容器中&#xff0c;通过反射实例化…...

带虚继承的类对象模型

文章目录 1、代码2、 单个虚继承3、vbptr是什么4、虚继承的多继承 1、代码 #include<iostream> using namespace std;class Base { public:int ma; };class Derive1 :virtual public Base { public:int mb; };class Derive2 :public Base { public:int mc; };class Deri…...

深度学习中的离群值

文章目录 深度学习中有离群值吗&#xff1f;深度学习中的离群值来源&#xff1a;处理离群值的策略&#xff1a;1. 数据预处理阶段&#xff1a;2. 数据增强和鲁棒模型&#xff1a;3. 模型训练阶段&#xff1a;4. 异常检测集成模型&#xff1a; 如何处理对抗样本&#xff1f;总结…...

如何利用Logo设计免费生成器创建专业级Logo

在当今的商业世界中&#xff0c;一个好的Logo是品牌身份的象征&#xff0c;它承载着公司的形象与理念。设计一个专业级的Logo不再需要花费大量的金钱和时间&#xff0c;尤其是当我们拥有Logo设计免费生成器这样的工具时。接下来&#xff0c;让我们深入探讨如何利用这些工具来创…...

Mysql SQL 超实用的7个日期算术运算实例(10k)

文章目录 前言1. 加上或减去若干天、若干月或若干年基本语法使用场景注意事项运用实例分析说明2. 确定两个日期相差多少天基本语法使用场景注意事项运用实例分析说明3. 确定两个日期之间有多少个工作日基本语法使用场景注意事项运用实例分析说明4. 确定两个日期相隔多少个月或多…...

运算指令(PLC)

加 ADD 减 SUB 乘 MUL 除 DIV 浮点运算 整数运算...

「Mac畅玩鸿蒙与硬件49」UI互动应用篇26 - 数字填色游戏

本篇教程将带你实现一个数字填色小游戏&#xff0c;通过简单的交互逻辑&#xff0c;学习如何使用鸿蒙开发组件创建趣味性强的应用。 关键词 UI互动应用数字填色动态交互逻辑判断游戏开发 一、功能说明 数字填色小游戏包含以下功能&#xff1a; 数字选择&#xff1a;用户点击…...

机器学习经典算法——逻辑回归

目录 算法介绍 算法概念 算法的优缺点 LogisticRegression()函数理解 环境准备 算法练习 算法介绍 算法概念 逻辑回归&#xff08;Logistic Regression&#xff09;是一种广泛应用于分类问题的机器学习算法。 它基于线性回归的思想&#xff0c;但通过引入一个逻辑函数&…...

【数据仓库金典面试题】—— 包含详细解答

大家好&#xff0c;我是摇光~&#xff0c;用大白话讲解所有你难懂的知识点 该篇面试题主要针对面试涉及到数据仓库的数据岗位。 以下都是经典的关于数据仓库的问题&#xff0c;希望对大家面试有用~ 1、什么是数据仓库&#xff1f;它与传统数据库有何区别&#xff1f; 数据仓库…...

【UE5 C++课程系列笔记】19——通过GConfig读写.ini文件

步骤 1. 新建一个Actor类&#xff0c;这里命名为“INIActor” 2. 新建一个配置文件“Test.ini” 添加一个自定义配置项 3. 接下来我们在“INIActor”类中获取并修改“CustomInt”的值。这里定义一个方法“GetINIVariable” 方法实现如下&#xff0c;其中第16行代码用于构建配…...

React 第五十五节 Router 中 useAsyncError的使用详解

前言 useAsyncError 是 React Router v6.4 引入的一个钩子&#xff0c;用于处理异步操作&#xff08;如数据加载&#xff09;中的错误。下面我将详细解释其用途并提供代码示例。 一、useAsyncError 用途 处理异步错误&#xff1a;捕获在 loader 或 action 中发生的异步错误替…...

Vue记事本应用实现教程

文章目录 1. 项目介绍2. 开发环境准备3. 设计应用界面4. 创建Vue实例和数据模型5. 实现记事本功能5.1 添加新记事项5.2 删除记事项5.3 清空所有记事 6. 添加样式7. 功能扩展&#xff1a;显示创建时间8. 功能扩展&#xff1a;记事项搜索9. 完整代码10. Vue知识点解析10.1 数据绑…...

服务器硬防的应用场景都有哪些?

服务器硬防是指一种通过硬件设备层面的安全措施来防御服务器系统受到网络攻击的方式&#xff0c;避免服务器受到各种恶意攻击和网络威胁&#xff0c;那么&#xff0c;服务器硬防通常都会应用在哪些场景当中呢&#xff1f; 硬防服务器中一般会配备入侵检测系统和预防系统&#x…...

从零开始打造 OpenSTLinux 6.6 Yocto 系统(基于STM32CubeMX)(九)

设备树移植 和uboot设备树修改的内容同步到kernel将设备树stm32mp157d-stm32mp157daa1-mx.dts复制到内核源码目录下 源码修改及编译 修改arch/arm/boot/dts/st/Makefile&#xff0c;新增设备树编译 stm32mp157f-ev1-m4-examples.dtb \stm32mp157d-stm32mp157daa1-mx.dtb修改…...

【决胜公务员考试】求职OMG——见面课测验1

2025最新版&#xff01;&#xff01;&#xff01;6.8截至答题&#xff0c;大家注意呀&#xff01; 博主码字不易点个关注吧,祝期末顺利~~ 1.单选题(2分) 下列说法错误的是:&#xff08; B &#xff09; A.选调生属于公务员系统 B.公务员属于事业编 C.选调生有基层锻炼的要求 D…...

Java面试专项一-准备篇

一、企业简历筛选规则 一般企业的简历筛选流程&#xff1a;首先由HR先筛选一部分简历后&#xff0c;在将简历给到对应的项目负责人后再进行下一步的操作。 HR如何筛选简历 例如&#xff1a;Boss直聘&#xff08;招聘方平台&#xff09; 直接按照条件进行筛选 例如&#xff1a…...

苹果AI眼镜:从“工具”到“社交姿态”的范式革命——重新定义AI交互入口的未来机会

在2025年的AI硬件浪潮中,苹果AI眼镜(Apple Glasses)正在引发一场关于“人机交互形态”的深度思考。它并非简单地替代AirPods或Apple Watch,而是开辟了一个全新的、日常可接受的AI入口。其核心价值不在于功能的堆叠,而在于如何通过形态设计打破社交壁垒,成为用户“全天佩戴…...

springboot 日志类切面,接口成功记录日志,失败不记录

springboot 日志类切面&#xff0c;接口成功记录日志&#xff0c;失败不记录 自定义一个注解方法 import java.lang.annotation.ElementType; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target;/***…...

Neko虚拟浏览器远程协作方案:Docker+内网穿透技术部署实践

前言&#xff1a;本文将向开发者介绍一款创新性协作工具——Neko虚拟浏览器。在数字化协作场景中&#xff0c;跨地域的团队常需面对实时共享屏幕、协同编辑文档等需求。通过本指南&#xff0c;你将掌握在Ubuntu系统中使用容器化技术部署该工具的具体方案&#xff0c;并结合内网…...

边缘计算网关提升水产养殖尾水处理的远程运维效率

一、项目背景 随着水产养殖行业的快速发展&#xff0c;养殖尾水的处理成为了一个亟待解决的环保问题。传统的尾水处理方式不仅效率低下&#xff0c;而且难以实现精准监控和管理。为了提升尾水处理的效果和效率&#xff0c;同时降低人力成本&#xff0c;某大型水产养殖企业决定…...