Flume学习笔记(3)—— Flume 自定义组件

前置知识:

Flume学习笔记(1)—— Flume入门-CSDN博客

Flume学习笔记(2)—— Flume进阶-CSDN博客

Flume 自定义组件

自定义 Interceptor

需求分析:使用 Flume 采集服务器本地日志,需要按照日志类型的不同,将不同种类的日志发往不同的分析系统

需要使用Flume 拓扑结构中的 Multiplexing 结构,Multiplexing的原理是,根据 event 中 Header 的某个 key 的值,将不同的 event 发送到不同的 Channel中,所以我们需要自定义一个 Interceptor,为不同类型的 event 的 Header 中的 key 赋予不同的值

实现流程:

代码

导入依赖:

<dependencies><dependency><groupId>org.apache.flume</groupId><artifactId>flume-ng-core</artifactId><version>1.9.0</version></dependency>
</dependencies>

自定义拦截器的代码:

package com.why.interceptor;import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;import java.util.ArrayList;
import java.util.List;
import java.util.Map;public class TypeInterceptor implements Interceptor {//存放事件的集合private List<Event> addHeaderEvents;@Overridepublic void initialize() {//初始化集合addHeaderEvents = new ArrayList<>();}//单个事件拦截@Overridepublic Event intercept(Event event) {//获取头信息Map<String, String> headers = event.getHeaders();//获取body信息String body = new String(event.getBody());//根据数据中是否包含”why“来分组if(body.contains("why")){headers.put("type","first");}else {headers.put("type","second");}return event;}//批量事件拦截@Overridepublic List<Event> intercept(List<Event> events) {//清空集合addHeaderEvents.clear();//遍历eventsfor(Event event : events){//给每一个事件添加头信息addHeaderEvents.add(intercept(event));}return addHeaderEvents;}@Overridepublic void close() {}//构建生成器public static class TypeBuilder implements Interceptor.Builder{@Overridepublic Interceptor build() {return new TypeInterceptor();}@Overridepublic void configure(Context context) {}}}

将代码打包放入flume安装路径下的lib文件夹中

配置文件

在job文件夹下创建group4目录,添加配置文件;

为 hadoop102 上的 Flume1 配置 1 个 netcat source,1 个 sink group(2 个 avro sink),并配置相应的 ChannelSelector 和 interceptor

# Name the components on this agent
a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1 c2# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = com.why.interceptor.TypeInterceptor$TypeBuilder
a1.sources.r1.selector.type = multiplexing
a1.sources.r1.selector.header = type
a1.sources.r1.selector.mapping.first = c1
a1.sources.r1.selector.mapping.second = c2# Describe the sink
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = hadoop103
a1.sinks.k1.port = 4141
a1.sinks.k2.type=avro
a1.sinks.k2.hostname = hadoop104
a1.sinks.k2.port = 4242# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Use a channel which buffers events in memory
a1.channels.c2.type = memory
a1.channels.c2.capacity = 1000
a1.channels.c2.transactionCapacity = 100# Bind the source and sink to the channel
a1.sources.r1.channels = c1 c2
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c2

hadoop103:配置一个 avro source 和一个 logger sink

a1.sources = r1
a1.sinks = k1
a1.channels = c1
a1.sources.r1.type = avro
a1.sources.r1.bind = hadoop103
a1.sources.r1.port = 4141
a1.sinks.k1.type = logger
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
a1.sinks.k1.channel = c1
a1.sources.r1.channels = c1

hadoop104:配置一个 avro source 和一个 logger sink

a1.sources = r1
a1.sinks = k1
a1.channels = c1
a1.sources.r1.type = avro
a1.sources.r1.bind = hadoop104
a1.sources.r1.port = 4242
a1.sinks.k1.type = logger
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
a1.sinks.k1.channel = c1
a1.sources.r1.channels = c1

执行指令

hadoop102:bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group4/flume-interceptor-flume.conf

hadoop103:bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group4/flume1-flume-logger.conf -Dflume.root.logger=INFO,console

hadoop104:bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group4/flume2-flume-logger.conf -Dflume.root.logger=INFO,console

然后hadoop102通过nc连接44444端口,发送数据:

在hadoop103和104上分别接收到:

自定义 Source

官方提供的文档:Flume 1.11.0 Developer Guide — Apache Flume

给出的示例代码如下:

public class MySource extends AbstractSource implements Configurable, PollableSource {private String myProp;@Overridepublic void configure(Context context) {String myProp = context.getString("myProp", "defaultValue");// Process the myProp value (e.g. validation, convert to another type, ...)// Store myProp for later retrieval by process() methodthis.myProp = myProp;}@Overridepublic void start() {// Initialize the connection to the external client}@Overridepublic void stop () {// Disconnect from external client and do any additional cleanup// (e.g. releasing resources or nulling-out field values) ..}@Overridepublic Status process() throws EventDeliveryException {Status status = null;try {// This try clause includes whatever Channel/Event operations you want to do// Receive new dataEvent e = getSomeData();// Store the Event into this Source's associated Channel(s)getChannelProcessor().processEvent(e);status = Status.READY;} catch (Throwable t) {// Log exception, handle individual exceptions as neededstatus = Status.BACKOFF;// re-throw all Errorsif (t instanceof Error) {throw (Error)t;}} finally {txn.close();}return status;}
}

需要继承AbstractSource,实现Configurable, PollableSource

实战需求分析

使用 flume 接收数据,并给每条数据添加前缀,输出到控制台。前缀可从 flume 配置文件中配置

代码

package com.why.source;import org.apache.flume.Context;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.source.AbstractSource;import java.util.HashMap;
import java.util.concurrent.ConcurrentMap;public class MySource extends AbstractSource implements PollableSource, Configurable {//定义配置文件将来要读取的字段private Long delay;private String field;//获取数据封装成 event 并写入 channel,这个方法将被循环调用@Overridepublic Status process() throws EventDeliveryException {try {//事件头信息HashMap<String,String> headerMap = new HashMap<>();//创建事件SimpleEvent event = new SimpleEvent();//循环封装事件for (int i = 0; i < 5; i++) {//设置头信息event.setHeaders(headerMap);//设置事件内容event.setBody((field + i).getBytes());//将事件写入ChannelgetChannelProcessor().processEvent(event);Thread.sleep(delay);}}catch (InterruptedException e) {throw new RuntimeException(e);}return Status.READY;}//backoff 步长@Overridepublic long getBackOffSleepIncrement() {return 0;}//backoff 最长时间@Overridepublic long getMaxBackOffSleepInterval() {return 0;}//初始化 context(读取配置文件内容)@Overridepublic void configure(Context context) {delay = context.getLong("delay");field = context.getString("field","Hello");}}

打包放到flume安装路径下的lib文件夹中;

配置文件

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1# Describe/configure the source
a1.sources.r1.type = com.why.source.MySource
a1.sources.r1.delay = 1000
a1.sources.r1.field = why# Describe the sink
a1.sinks.k1.type = logger# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

执行指令

hadoop102上:bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group5/mysource.conf -Dflume.root.logger=INFO,console

结果如下:

自定义 Sink

Sink 不断地轮询 Channel 中的事件且批量地移除它们,并将这些事件批量写入到存储或索引系统、或者被发送到另一个 Flume Agent。

Sink 是完全事务性的。在从 Channel 批量删除数据之前,每个 Sink 用 Channel 启动一个事务。批量事件一旦成功写出到存储系统或下一个 Flume Agent,Sink 就利用 Channel 提交事务。事务一旦被提交,该 Channel 从自己的内部缓冲区删除事件

官方文档:Flume 1.11.0 Developer Guide — Apache Flume

接口实例:

public class MySink extends AbstractSink implements Configurable {private String myProp;@Overridepublic void configure(Context context) {String myProp = context.getString("myProp", "defaultValue");// Process the myProp value (e.g. validation)// Store myProp for later retrieval by process() methodthis.myProp = myProp;}@Overridepublic void start() {// Initialize the connection to the external repository (e.g. HDFS) that// this Sink will forward Events to ..}@Overridepublic void stop () {// Disconnect from the external respository and do any// additional cleanup (e.g. releasing resources or nulling-out// field values) ..}@Overridepublic Status process() throws EventDeliveryException {Status status = null;// Start transactionChannel ch = getChannel();Transaction txn = ch.getTransaction();txn.begin();try {// This try clause includes whatever Channel operations you want to doEvent event = ch.take();// Send the Event to the external repository.// storeSomeData(e);txn.commit();status = Status.READY;} catch (Throwable t) {txn.rollback();// Log exception, handle individual exceptions as neededstatus = Status.BACKOFF;// re-throw all Errorsif (t instanceof Error) {throw (Error)t;}}return status;}
}

自定义MySink 需要继承 AbstractSink 类并实现 Configurable 接口

实战需求分析

使用 flume 接收数据,并在 Sink 端给每条数据添加前缀和后缀,输出到控制台。前后缀可在 flume 任务配置文件中配置

代码

package com.why.sink;import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;public class MySink extends AbstractSink implements Configurable {//创建 Logger 对象private static final Logger LOG = LoggerFactory.getLogger(AbstractSink.class);//前后缀private String prefix;private String suffix;@Overridepublic Status process() throws EventDeliveryException {//声明返回值状态信息Status status;//获取当前 Sink 绑定的 ChannelChannel ch = getChannel();//获取事务Transaction txn = ch.getTransaction();//声明事件Event event;//开启事务txn.begin();//读取 Channel 中的事件,直到读取到事件结束循环while (true) {event = ch.take();if (event != null) {break;}}try {//处理事件(打印)LOG.info(prefix + new String(event.getBody()) + suffix);//事务提交txn.commit();status = Status.READY;} catch (Exception e) {//遇到异常,事务回滚txn.rollback();status = Status.BACKOFF;} finally {//关闭事务txn.close();}return status;}@Overridepublic void configure(Context context) {prefix = context.getString("prefix", "hello");suffix = context.getString("suffix");}
}

打包放到flume安装路径下的lib文件夹中;

配置文件

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444# Describe the sink
a1.sinks.k1.type = com.why.sink.MySink
a1.sinks.k1.prefix = why:
a1.sinks.k1.suffix = :why# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

执行指令

hadoop102上:bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group6/mysink.conf -Dflume.root.logger=INFO,console

结果如下:

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

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

相关文章

数字化未来:实时云渲染在智慧城市中的创新应用

数字中国战略"是国家推动数字经济发展的战略框架。这个战略旨在加速数字化转型&#xff0c;推动信息技术在各个领域的应用&#xff0c;提高社会经济效益和人民生活质量。而智慧城市作为其中的重要一环&#xff0c;重要性不言而喻。 智慧城市是当今城市发展的热点和趋势&a…

第十一周任务总结

本周任务总结 本周物联网方面主要继续进行网关的二次开发与规则引擎实现设备联动的实现 非物联网方面主要复习了docker的使用与算法的学习 1.网关的二次开发&#xff0c;本周将实现debug调试输出的文件下载到了网关&#xff0c;但网关出了问题无法连接&#xff0c;最终跟客服…

Libvirt-Qemu-Kvm 操作手记

(持续更新~) 本文主要用于记录在操作libvirt qemu kvm过程中遇到的问题及原因分析。 Hugepage 让qemu使用大页可以减少tdp的size&#xff0c;一定程度上可以提高性能&#xff1b;使用大页可以用memfd或者file backend。 memfd 操作步骤如下&#xff1a; 在系统中reserv…

数据结构 堆

手写堆&#xff0c;而非stl中的堆 如何手写一个堆&#xff1f; //将数组建成堆 <O(n) for (int i n / 2;i;i--) //从n/2开始down down(i); 从n/2元素开始down&#xff0c;最下面一层元素的个数是n/2&#xff0c;其余上面的元素的个数是n/2&#xff0c;从最下面一层到最高层…

【四元数简述】

w cos(theta/2) x ax * sin(theta/2) y ay * sin(theta/2) z az * sin(theta/2) 向量(x,y,z)是旋转轴 a 是任意正数 theta是旋转角度。 上面就是一个四元数表示旋转。 如何使用 空间中向量&#xff08;1,2,3&#xff09;扩展为&#xff08;0,1,2,3&#…

electron使用better-sqlite3打包失败(electron打包有进程没有界面)

remove *\chrome_100_percent.pak: Access is denied. 解决&#xff1a; 管理员权限执行&#xff1a;taskkill /IM 你的进程名.exe /F&#xff0c;再次执行build electron使用better-sqlite3打包后有进程没有界面 原因是代码及依赖包安装有误&#xff0c;模块丢失。主要分享的…

Web前端—移动Web第二天(空间转换、动画、综合案例:全名出游)

版本说明 当前版本号[20231118]。 版本修改说明20231118初版 目录 文章目录 版本说明目录移动 Web 第二天01-空间转换空间转换简介平移视距旋转左手法则rotate3d-了解立体呈现案例-3d导航缩放 02-动画动画实现步骤animation复合属性animation拆分写法案例-走马灯精灵动画多组…

分享个pyppeteer 通用爬虫

突然发现自己之前写过的代码 虽然现在爬虫都逆向了&#xff0c;但是有时候获取个一次性数据&#xff0c;或者不需要维护的时候 pyppeteer 还是非常好用的。 因为pyppeteer 是puppeteer的python 版本 所以不是特别火 网上资料也非常少 .简单记录一下 如果想详细了解https://z…

某大型房地产公司绩效面谈项目成功案例纪实

——开展有效的绩效面谈&#xff0c;促进和完善管理工作 【客户行业】房地产行业&#xff1b;国有企业 【问题类型】绩效面谈改进 【客户背景】 某大型房地产公司是某国企集团的省级分公司&#xff0c;集团公司现拥有北京、上海、广州、山东等8大区域公司&#xff0c;现有员…

每天一道算法题(六)——返回一组数字中所有和为 0 且不重复的三元组

文章目录 前言1、问题2、示例3、解决方法4、效果5、注意点 前言 注意&#xff1a;答案中不可以包含重复的三元组。 1、问题 给你一个整数数组 nums &#xff0c;判断是否存在三元组 [nums[i], nums[j], nums[k]] 满足 i ! j、i ! k 且 j ! k &#xff0c;同时还满足 nums[i] n…

WMS仓库管理系统库位分配规划

分配说明 每个商品必须至少有一个分拣位&#xff08;整箱或拆零均可&#xff09;如果只设了拆零分拣位&#xff0c;则入库不分配存储位&#xff0c;只上拆零分拣位&#xff0c;对于DPS分拣商品是一品固定一个货位&#xff0c;对于RF分拣商品是动态分配货位&#xff0c;一个商品…

记录我常用的免费API接口

目录 1.随机中英文句子 2.随机中英文句子&#xff08;带图片和音频&#xff09; 3.随机一句诗 4.随机一句话 5.随机一句情话 6. 随机一句舔狗语录 7.历史上的今天 8.获取来访者ip地址 9&#xff1a;获取手机号信息 10. 垃圾分类查询 11.字典查询 12.QQ信息查询 1.随…

golang学习笔记——条件表达式

文章目录 条件表达式if 语句的语法复合 if 语句 条件表达式 在任何编程语言中&#xff0c;最基本的控制流都是 if/else 语句。 在 Go 中&#xff0c;if/else 语句非常简单。 但是&#xff0c;你需要先了解一些差异&#xff0c;然后才能得心应手地编写 Go 程序。 让我们看看 i…

JAVA Web应用常见漏洞与修复建议

背景&#xff1a; 在工作中&#xff0c;我们的交付团队在交付项目时&#xff0c;可能会遇到甲方会使用一些第三方工具&#xff08;奇安信等&#xff09;对项目代码进行扫描&#xff0c;特别是一些对安全性要求比较高的企业&#xff0c;比如涉及到一些证券公司、银行、金融等。他…

【网络学习笔记】

记录一下关于域名&#xff0c;dns&#xff0c;反向代理知识的一些内容 通过阿里云函数进行反向代理 阿里云函数计算fanxiangdaili 逻辑 和cloudflare的workers的差不多&#xff08;前几天突然不能用了&#xff0c;使用魔法还能用&#xff0c;不过今天又莫名其妙恢复了&#xf…

基础组件-流量回放(全链路流量回放预研)

一、调研 日常测试工作中&#xff0c;遇到问题如下&#xff1a; 1、服务架构升级或者重构&#xff0c;需要对原有的接口做全量回归 2、对于业务逻辑复杂的场景&#xff0c;每个迭代版本都需要大量的时间进行回归测试 3、编写自动化用例时复杂场景造数麻烦&#xff0c;日常自动…

Mybatis-Plus《学习笔记 22版尚硅谷 》——感谢【尚硅谷】官方文档

Mybatis-Plus《学习笔记 22版尚硅谷 》 一、MyBatis-Plus1.简介2.特性3.支持数据库4.框架结构5.官方地址 二、入门案例1.开发环境2.建库建表3.创建工程4.配置编码5.测试查询 三、增删改查1.BaseMapper<T>2.调用Mapper层实现CRUD2.1 插入2.2 删除a、根据ID删除数据b、根据…

单元测试实战(三)JPA 的测试

为鼓励单元测试&#xff0c;特分门别类示例各种组件的测试代码并进行解说&#xff0c;供开发人员参考。 本文中的测试均基于JUnit5。 单元测试实战&#xff08;一&#xff09;Controller 的测试 单元测试实战&#xff08;二&#xff09;Service 的测试 单元测试实战&am…

Node.js之TCP(net)

Hi I’m Shendi Node.js之TCP&#xff08;net&#xff09; 最近使用Nodejs编写程序&#xff0c;需要用到自己编写的分布式工具&#xff0c;于是需要将Java版的用NodeJs重新写一遍&#xff0c;需要使用到TCP通信&#xff0c;于是在这里记录下Node.js TCP 的使用方法 依赖 需要使…

【Python进阶】近200页md文档14大体系知识点,第4篇:linux命令和vim使用

本文从14大模块展示了python高级用的应用。分别有Linux命令&#xff0c;多任务编程、网络编程、Http协议和静态Web编程、htmlcss、JavaScript、jQuery、MySql数据库的各种用法、python的闭包和装饰器、mini-web框架、正则表达式等相关文章的详细讲述。 全套Python进阶笔记地址…