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;}}
}
拦截器实现
- 继承Interceptor接口
- 实现单event处理
- 实现批量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 学习(一):Combine 初印象Swift Combine 学习(二):发布者 PublisherSwift Combine 学习(三):Subscription和 SubscriberSwift Combine 学习(四&…...
leetcode 173.二叉搜索树迭代器栈绝妙思路
以上算法题中一个比较好的实现思路就是利用栈来进行实现,以下方法三就是利用栈来进行实现的,思路很好,很简练。进行next的时候,先是一直拿到左边的子树,直到null为止,这一步比较好思考一点,下一…...
df.groupby([pd.Grouper(freq=‘1M‘, key=‘Date‘), ‘Buyer‘]).sum()
df.groupby([pd.Grouper(freq1M, keyDate), Buyer]).sum() 用于根据特定的时间频率和买家(Buyer)对 DataFrame 进行分组,然后计算每个分组的总和。下面是对这行代码的逐步解释: df.groupby([...]):这个操作会根据传入的…...
LLM - 使用 LLaMA-Factory 部署大模型 HTTP 多模态服务 (4)
欢迎关注我的CSDN:https://spike.blog.csdn.net/ 本文地址:https://spike.blog.csdn.net/article/details/144881432 大模型的 HTTP 服务,通过网络接口,提供 AI 模型功能的服务,允许通过发送 HTTP 请求,交互…...
icp备案网站个人备案与企业备案的区别
个人备案和企业备案是在进行ICP备案时需要考虑的两种不同情况。个人备案是指个人拥有的网站进行备案,而企业备案则是指企业或组织名下的网站进行备案。这两者在备案过程中有一些明显的区别。 首先,个人备案相对来说流程较为简单。个人备案只需要提供个人…...
如何不修改模型参数来强化大语言模型 (LLM) 能力?
前言 如果你对这篇文章感兴趣,可以点击「【访客必读 - 指引页】一文囊括主页内所有高质量博客」,查看完整博客分类与对应链接。 大语言模型 (Large Language Model, LLM, e.g. ChatGPT) 的参数量少则几十亿,多则上千亿,对其的训…...
AF3 AtomAttentionEncoder类的init_pair_repr方法解读
AlphaFold3 的 AtomAttentionEncoder 类中,init_pair_repr 方法方法负责为原子之间的关系计算成对表示(pair representation),这是原子转变器(atom transformer)模型的关键组成部分,直接影响对蛋白质/分子相互作用的建模。 init_pair_repr源代码: def init_pair_repr(…...
DDoS攻击防御方案大全
1. 引言 随着互联网的迅猛发展,DDoS(分布式拒绝服务)攻击成为了网络安全领域中最常见且危害严重的攻击方式之一。DDoS攻击通过向目标网络或服务发送大量流量,导致服务器过载,最终使其无法响应合法用户的请求。本文将深…...
Vue中常用指令
一、内容渲染指令 1.v-text:操作纯文本,用于更新标签包含的文本,但是使用不灵活,无法拼接字符串,会覆盖文本,可以简写为{{}},{{}}支持逻辑运算。 用法示例: //把name对应的值渲染到…...
Servlet解析
概念 Servlet是运行在服务端的小程序(Server Applet),可以处理客户端的请求并返回响应,主要用于构建动态的Web应用,是SpringMVC的基础。 生命周期 加载和初始化 默认在客户端第一次请求加载到容器中,通过反射实例化…...
带虚继承的类对象模型
文章目录 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…...
深度学习中的离群值
文章目录 深度学习中有离群值吗?深度学习中的离群值来源:处理离群值的策略:1. 数据预处理阶段:2. 数据增强和鲁棒模型:3. 模型训练阶段:4. 异常检测集成模型: 如何处理对抗样本?总结…...
如何利用Logo设计免费生成器创建专业级Logo
在当今的商业世界中,一个好的Logo是品牌身份的象征,它承载着公司的形象与理念。设计一个专业级的Logo不再需要花费大量的金钱和时间,尤其是当我们拥有Logo设计免费生成器这样的工具时。接下来,让我们深入探讨如何利用这些工具来创…...
Mysql SQL 超实用的7个日期算术运算实例(10k)
文章目录 前言1. 加上或减去若干天、若干月或若干年基本语法使用场景注意事项运用实例分析说明2. 确定两个日期相差多少天基本语法使用场景注意事项运用实例分析说明3. 确定两个日期之间有多少个工作日基本语法使用场景注意事项运用实例分析说明4. 确定两个日期相隔多少个月或多…...
运算指令(PLC)
加 ADD 减 SUB 乘 MUL 除 DIV 浮点运算 整数运算...
「Mac畅玩鸿蒙与硬件49」UI互动应用篇26 - 数字填色游戏
本篇教程将带你实现一个数字填色小游戏,通过简单的交互逻辑,学习如何使用鸿蒙开发组件创建趣味性强的应用。 关键词 UI互动应用数字填色动态交互逻辑判断游戏开发 一、功能说明 数字填色小游戏包含以下功能: 数字选择:用户点击…...
机器学习经典算法——逻辑回归
目录 算法介绍 算法概念 算法的优缺点 LogisticRegression()函数理解 环境准备 算法练习 算法介绍 算法概念 逻辑回归(Logistic Regression)是一种广泛应用于分类问题的机器学习算法。 它基于线性回归的思想,但通过引入一个逻辑函数&…...
【数据仓库金典面试题】—— 包含详细解答
大家好,我是摇光~,用大白话讲解所有你难懂的知识点 该篇面试题主要针对面试涉及到数据仓库的数据岗位。 以下都是经典的关于数据仓库的问题,希望对大家面试有用~ 1、什么是数据仓库?它与传统数据库有何区别? 数据仓库…...
【UE5 C++课程系列笔记】19——通过GConfig读写.ini文件
步骤 1. 新建一个Actor类,这里命名为“INIActor” 2. 新建一个配置文件“Test.ini” 添加一个自定义配置项 3. 接下来我们在“INIActor”类中获取并修改“CustomInt”的值。这里定义一个方法“GetINIVariable” 方法实现如下,其中第16行代码用于构建配…...
云计算——弹性云计算器(ECS)
弹性云服务器:ECS 概述 云计算重构了ICT系统,云计算平台厂商推出使得厂家能够主要关注应用管理而非平台管理的云平台,包含如下主要概念。 ECS(Elastic Cloud Server):即弹性云服务器,是云计算…...
【Java学习笔记】Arrays类
Arrays 类 1. 导入包:import java.util.Arrays 2. 常用方法一览表 方法描述Arrays.toString()返回数组的字符串形式Arrays.sort()排序(自然排序和定制排序)Arrays.binarySearch()通过二分搜索法进行查找(前提:数组是…...
cf2117E
原题链接:https://codeforces.com/contest/2117/problem/E 题目背景: 给定两个数组a,b,可以执行多次以下操作:选择 i (1 < i < n - 1),并设置 或,也可以在执行上述操作前执行一次删除任意 和 。求…...
镜像里切换为普通用户
如果你登录远程虚拟机默认就是 root 用户,但你不希望用 root 权限运行 ns-3(这是对的,ns3 工具会拒绝 root),你可以按以下方法创建一个 非 root 用户账号 并切换到它运行 ns-3。 一次性解决方案:创建非 roo…...
从零开始打造 OpenSTLinux 6.6 Yocto 系统(基于STM32CubeMX)(九)
设备树移植 和uboot设备树修改的内容同步到kernel将设备树stm32mp157d-stm32mp157daa1-mx.dts复制到内核源码目录下 源码修改及编译 修改arch/arm/boot/dts/st/Makefile,新增设备树编译 stm32mp157f-ev1-m4-examples.dtb \stm32mp157d-stm32mp157daa1-mx.dtb修改…...
相机Camera日志分析之三十一:高通Camx HAL十种流程基础分析关键字汇总(后续持续更新中)
【关注我,后续持续新增专题博文,谢谢!!!】 上一篇我们讲了:有对最普通的场景进行各个日志注释讲解,但相机场景太多,日志差异也巨大。后面将展示各种场景下的日志。 通过notepad++打开场景下的日志,通过下列分类关键字搜索,即可清晰的分析不同场景的相机运行流程差异…...
EtherNet/IP转DeviceNet协议网关详解
一,设备主要功能 疆鸿智能JH-DVN-EIP本产品是自主研发的一款EtherNet/IP从站功能的通讯网关。该产品主要功能是连接DeviceNet总线和EtherNet/IP网络,本网关连接到EtherNet/IP总线中做为从站使用,连接到DeviceNet总线中做为从站使用。 在自动…...
初学 pytest 记录
安装 pip install pytest用例可以是函数也可以是类中的方法 def test_func():print()class TestAdd: # def __init__(self): 在 pytest 中不可以使用__init__方法 # self.cc 12345 pytest.mark.api def test_str(self):res add(1, 2)assert res 12def test_int(self):r…...
从面试角度回答Android中ContentProvider启动原理
Android中ContentProvider原理的面试角度解析,分为已启动和未启动两种场景: 一、ContentProvider已启动的情况 1. 核心流程 触发条件:当其他组件(如Activity、Service)通过ContentR…...
Python常用模块:time、os、shutil与flask初探
一、Flask初探 & PyCharm终端配置 目的: 快速搭建小型Web服务器以提供数据。 工具: 第三方Web框架 Flask (需 pip install flask 安装)。 安装 Flask: 建议: 使用 PyCharm 内置的 Terminal (模拟命令行) 进行安装,避免频繁切换。 PyCharm Terminal 配置建议: 打开 Py…...
