Flume学习笔记（2）—— Flume进阶

Flume进阶

Flume 事务

事务处理流程如下：

Put

  • doPut：将批数据先写入临时缓冲区putList
  • doCommit：检查channel内存队列是否足够合并。
  • doRollback：channel内存队列空间不足，回滚数据

Take

  • doTake：将数据取到临时缓冲区takeList，并将数据发送到HDFS
  • doCommit：如果数据全部发送成功，则清除临时缓冲区takeList
  • doRollback：数据发送过程中如果出现异常，rollback将临时缓冲区takeList中的数据归还给channel内存队列

Flume Agent 内部原理

ChannelSelector

ChannelSelector 的作用就是选出 Event 将要被发往哪个 Channel

其共有两种类型，分别是 Replicating（复制）和 Multiplexing（多路复用）

  • ReplicatingSelector 会将同一个 Event 发往所有的 Channel
  • Multiplexing 会根据相应的原则，将不同的 Event 发往不同的 Channel

SinkProcessor

SinkProcessor 共有三种类型 ， 分别是 DefaultSinkProcessor 、LoadBalancingSinkProcessor、FailoverSinkProcessor

  • DefaultSinkProcessor 对应的是单个的Sink
  • LoadBalancingSinkProcessor 可以实现负载均衡的功能
  • FailoverSinkProcessor 可以实现错误恢复的功能

Flume 拓扑结构

简单串联

将多个 flume 顺序连接起来，从最初的 source 开始到最终 sink 传送的目的存储系统

不建议桥接过多的 flume 数量， flume 数量过多不仅会影响传输速率，而且一旦传输过程中某个节点 flume 宕机，会影响整个传输系统

复制和多路复用

（单 source，多 channel、sink）

Flume 支持将事件流向一个或者多个目的地

这种模式可以将相同数据复制到多个channel 中，或者将不同数据分发到不同的 channel 中，sink 可以选择传送到不同的目的地

负载均衡和故障转移

Flume支持使用将多个sink逻辑上分到一个sink组，sink组配合不同的SinkProcessor可以实现负载均衡和错误恢复的功能、

这里的agent1有三个sink，分别连接agent2，agent3，agent4，即使其中有的sink出现了故障，数据还是能同步到hdfs

聚合

业务中常用，比如说日志采集功能：

日常 web 应用通常分布在上百个服务器，大者甚至上千个、上万个服务器，产生的日志处理起来也非常麻烦

可以采用聚合的方式，每台服务器部署一个 flume 采集日志，传送到一个集中收集日志的flume，再由此 flume 上传到 hdfs、hive、hbase 等，进行日志分析

Flume实战案例

复制和多路复用

需求：使用 Flume-1 监控文件变动

  1. Flume-1 将变动内容传递给 Flume-2，Flume-2 负责存储到 HDFS
  2. Flume-1 将变动内容传递给 Flume-3，Flume-3 负责输出到 Local FileSystem

实现流程：
1.在job下创建文件夹group1，并在其中创建配置文件flume-file-flume.conf

配置文件中需要有1个source，2个channel，2个sink

# Name the components on this agent
a1.sources = r1
a1.sinks = k1 k2
a1.channels = c1 c2# 将数据流复制给所有 channel
a1.sources.r1.selector.type = replicating# Describe/configure the source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /opt/module/apache-hive-3.1.2-bin/logs/hive.log
a1.sources.r1.shell = /bin/bash -c# Describe the sink
# sink 端的 avro 是一个数据发送者
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = hadoop102
a1.sinks.k1.port = 4141
a1.sinks.k2.type = avro
a1.sinks.k2.hostname = hadoop102
a1.sinks.k2.port = 4142# Describe the channel
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
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

该配置文件的作用是将数据发送到两个不同的sink，再由sink发送到其他的agent进行处理

2.创建配置文件flume-flume-hdfs.conf

# Name the components on this agent
a2.sources = r1
a2.sinks = k1
a2.channels = c1# Describe/configure the source
# source 端的 avro 是一个数据接收服务
a2.sources.r1.type = avro
a2.sources.r1.bind = hadoop102
a2.sources.r1.port = 4141# Describe the sink
a2.sinks.k1.type = hdfs
a2.sinks.k1.hdfs.path = hdfs://hadoop102:8020/flume2/%Y%m%d/%H
#上传文件的前缀
a2.sinks.k1.hdfs.filePrefix = flume2-
#是否按照时间滚动文件夹
a2.sinks.k1.hdfs.round = true
#多少时间单位创建一个新的文件夹
a2.sinks.k1.hdfs.roundValue = 1
#重新定义时间单位
a2.sinks.k1.hdfs.roundUnit = hour
#是否使用本地时间戳
a2.sinks.k1.hdfs.useLocalTimeStamp = true
#积攒多少个 Event 才 flush 到 HDFS 一次
a2.sinks.k1.hdfs.batchSize = 100
#设置文件类型，可支持压缩
a2.sinks.k1.hdfs.fileType = DataStream
#多久生成一个新的文件
a2.sinks.k1.hdfs.rollInterval = 30
#设置每个文件的滚动大小大概是 128M
a2.sinks.k1.hdfs.rollSize = 134217700
#文件的滚动与 Event 数量无关
a2.sinks.k1.hdfs.rollCount = 0# Describe the channel
a2.channels.c1.type = memory
a2.channels.c1.capacity = 1000
a2.channels.c1.transactionCapacity = 100# Bind the source and sink to the channel
a2.sources.r1.channels = c1
a2.sinks.k1.channel = c1

source绑定上一个agent的sink1，然后上传到hdfs

3.创建配置文件：flume-flume-dir.conf

# Name the components on this agent
a3.sources = r1
a3.sinks = k1
a3.channels = c2# Describe/configure the source
a3.sources.r1.type = avro
a3.sources.r1.bind = hadoop102
a3.sources.r1.port = 4142# Describe the sink
a3.sinks.k1.type = file_roll
a3.sinks.k1.sink.directory = /home/why/data/flumeDemo/test1# Describe the channel
a3.channels.c2.type = memory
a3.channels.c2.capacity = 1000
a3.channels.c2.transactionCapacity = 100# Bind the source and sink to the channel
a3.sources.r1.channels = c2
a3.sinks.k1.channel = c2

参数说明：

sink类型为file_roll:Flume 1.11.0 User Guide — Apache Flume

可以将events保存到本地文件系统

  • directory：本地文件系统保存数据的路径（注意，该路径必须已经存在才可以）

4.分别启动相应的flume进程：

nohup bin/flume-ng agent --conf conf/ --name a3 --conf-file job/group1/flume-flume-dir.conf &

nohup bin/flume-ng agent --conf conf/ --name a2 --conf-file job/group1/flume-flume-hdfs.conf &

nohup bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group1/flume-file-flume.conf &
5.在hdfs和文件夹中都能看到相应的内容：

hdfs：

文件系统：

负载均衡和故障转移

需求：使用 Flume1 监控一个端口，其 sink 组中的 sink 分别对接 Flume2 和 Flume3，采用FailoverSinkProcessor，实现故障转移的功能

实现流程：

1.在/opt/module/flume/job 目录下创建 group2 文件夹，创建配置文件flume-netcat-flume.conf

配置 1 个 netcat source 和 1 个 channel、1 个 sink group（2 个 sink），分别输送给flume-flume-console1 和 flume-flume-console2

# Name the components on this agent
a1.sources = r1
a1.channels = c1
a1.sinkgroups = g1
a1.sinkgroups.g1.processor.type = failover
a1.sinkgroups.g1.processor.priority.k1 = 5
a1.sinkgroups.g1.processor.priority.k2 = 10
a1.sinkgroups.g1.processor.maxpenalty = 10000
a1.sinks = k1 k2# 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 = avro
a1.sinks.k1.hostname = hadoop102
a1.sinks.k1.port = 4141
a1.sinks.k2.type = avro
a1.sinks.k2.hostname = hadoop102
a1.sinks.k2.port = 4142# Describe the channel
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.sinkgroups.g1.sinks = k1 k2
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c1

参数说明：Flume 1.11.0 User Guide — Apache Flume

通过sink groups在一个agent中定义多个sink，并可以配置sink processor使用：Flume 1.11.0 User Guide — Apache Flume

2.创建 flume-flume-console1.conf

# Name the components on this agent
a2.sources = r1
a2.sinks = k1
a2.channels = c1# Describe/configure the source
a2.sources.r1.type = avro
a2.sources.r1.bind = hadoop102
a2.sources.r1.port = 4141# Describe the sink
a2.sinks.k1.type = logger# Describe the channel
a2.channels.c1.type = memory
a2.channels.c1.capacity = 1000
a2.channels.c1.transactionCapacity = 100# Bind the source and sink to the channel
a2.sources.r1.channels = c1
a2.sinks.k1.channel = c1

sink输出到本地的控制台

3.创建 flume-flume-console2.conf

# Name the components on this agent
a3.sources = r1
a3.sinks = k1
a3.channels = c2# Describe/configure the source
a3.sources.r1.type = avro
a3.sources.r1.bind = hadoop102
a3.sources.r1.port = 4142# Describe the sink
a3.sinks.k1.type = logger# Describe the channel
a3.channels.c2.type = memory
a3.channels.c2.capacity = 1000
a3.channels.c2.transactionCapacity = 100# Bind the source and sink to the channel
a3.sources.r1.channels = c2
a3.sinks.k1.channel = c2

sink输出到本地的控制台

4.执行指令：

bin/flume-ng agent --conf conf/ --name a3 --conf-file job/group2/flume-flume-console2.conf -Dflume.root.logger=INFO,console

bin/flume-ng agent --conf conf/ --name a2 --conf-file job/group2/flume-flume-console1.conf -Dflume.root.logger=INFO,console

bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group2/flume-netcat-flume.conf

5.使用nc localhost 44444发送数据

由于console2设置的优先级高于console1，因此数据由console2接收到；

接下来将console2进程kill掉，数据就由console1接收了：

聚合

需求：

hadoop102 上的 Flume-1 监控文件/home/why/data/flumeDemo/test3/test3.log

hadoop103 上的 Flume-2 监控某一个端口的数据流

Flume-1 与 Flume-2 将数据发送给 hadoop104 上的 Flume-3，Flume-3 将最终数据打印到控制台

实现流程：

1.首先在三台服务器的job文件夹先创建目录group3

2.在hadoop102上，创建配置文件flume1-logger-flume.conf，source用于监控log日志文件，sink用于输出数据到下一级的Flume

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1# Describe/configure the source
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /home/why/data/flumeDemo/test3/test3.log
a1.sources.r1.shell = /bin/bash -c# Describe the sink
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = hadoop104
a1.sinks.k1.port = 4141# Describe the channel
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

3.在hadoop103上，创建配置文件flume2-netcat-flume.conf，source用于监控端口44444的数据流，sink用于将数据传输到下一级的flume

# Name the components on this agent
a2.sources = r1
a2.sinks = k1
a2.channels = c1# Describe/configure the source
a2.sources.r1.type = netcat
a2.sources.r1.bind = localhost
a2.sources.r1.port = 44444# Describe the sink
a2.sinks.k1.type = avro
a2.sinks.k1.hostname = hadoop104
a2.sinks.k1.port = 4141# Use a channel which buffers events in memory
a2.channels.c1.type = memory
a2.channels.c1.capacity = 1000
a2.channels.c1.transactionCapacity = 100# Bind the source and sink to the channel
a2.sources.r1.channels = c1
a2.sinks.k1.channel = c1

注意，这两个agent的sink目的地都是hadoop104这一个服务器，因此hostname和port都相同

4.在hadoop104上创建配置文件flume3-flume-logger.conf，source用于接收flume1和flume2发送来的数据流，sink用于输出数据到控制台；

# Name the components on this agent
a3.sources = r1
a3.sinks = k1
a3.channels = c1# Describe/configure the source
a3.sources.r1.type = avro
a3.sources.r1.bind = hadoop104
a3.sources.r1.port = 4141# Describe the sink
a3.sinks.k1.type = logger# Describe the channel
a3.channels.c1.type = memory
a3.channels.c1.capacity = 1000
a3.channels.c1.transactionCapacity = 100# Bind the source and sink to the channel
a3.sources.r1.channels = c1
a3.sinks.k1.channel = c1

5.分别在三台服务器上执行指令

hadoop104： bin/flume-ng agent --conf conf/ --name a3 --conf-file job/group3/flume3-flume-logger.conf -Dflume.root.logger=INFO,console

hadoop102：bin/flume-ng agent --conf conf/ --name a1 --conf-file job/group3/flume1-logger-flume.conf

hadoop103：bin/flume-ng agent --conf conf/ --name a2 --conf-file job/group3/flume2-netcat-flume.conf

6.在hadoop102上向日志文件中追加内容：

echo "hello" > /home/why/data/flumeDemo/test3/test3.log

在hadoop103中通过nc hadoop103 44444向44444端口发送数据；

然后在hadoop104中即可接收到数据：

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

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

相关文章

应用场景丨迭代市政综合管廊监测系统建设

应用场景丨迭代市政综合管廊监测系统建设

市政综合管廊是指在城市地下建造的隧道空间，将市政、电力、通讯、燃气、给排水等各种管线集于一体，实施统一规划、设计、建设和管理。综合管廊有利于解决反复开挖路面、架空线网密集、管线事故频发等问题，是保障城市运行的重要基础设施和“生…
阅读更多...
Dubbo的优雅下线原理分析

Dubbo的优雅下线原理分析

文/朱季谦 Dubbo如何实现优雅下线？ 这个问题困扰了我一阵，既然有优雅下线这种说法，那么，是否有非优雅下线的说法呢？ 这，还真有。 可以从linux进程关闭说起，其实，我们经常使用到杀…
阅读更多...
策略模式在数据接收和发送场景的应用（升级版）

策略模式在数据接收和发送场景的应用（升级版）

1.背景 在数据接收和发送场景打算使用了 if else 进行判断： if("A".equals(system)){ASystem.sync("向A同步数据"); } if("B".equals(system)){BSystem.sync("向B同步数据"); } ... 非常麻烦，需求多了很臃肿&…
阅读更多...
C++标准模板（STL）- 类型支持 （类型关系，检查两个类型是否相同，std::is_same）

C++标准模板（STL）- 类型支持 （类型关系，检查两个类型是否相同，std::is_same）

类型特性 类型特性 类型特性定义一个编译时基于模板的结构&#xff0c;以查询或修改类型的属性。 试图特化定义于 <type_traits> 头文件的模板导致未定义行为&#xff0c;除了 std::common_type 可依照其所描述特化。 定义于<type_traits>头文件的模板可以用不完…
阅读更多...
macos 配置ndk环境

macos 配置ndk环境

选择Android Studio下默认的ndk环境 mac电脑的ndk默认路径一般是 /Users/user_name/Library/Android/sdk/ndk/version_code 其中user_name为自己电脑的用户名&#xff0c;version_code为自己ndk安装的版本号&#xff0c;比如我这里电脑的ndk路径就是 /Users/zhangsan/Libra…
阅读更多...
多线程Thread（初阶一：认识线程）

多线程Thread（初阶一：认识线程）

目录 一、引用线程的原因 二、线程的概念 三、进程和线程的区别 四、多线程编程 一、引用线程的原因 多任务操作系统&#xff0c;希望系统能同时运行多个任务。所以会涉及到进程&#xff0c;需要对进程进行管理、调度等。 而单任务操作系统&#xff0c;就完全不涉及到进程…
阅读更多...
【漏洞复现】浙大恩特CRM文件上传0day

【漏洞复现】浙大恩特CRM文件上传0day

漏洞描述 浙大恩特客户资源管理系统任意文件上传漏洞 免责声明 技术文章仅供参考，任何个人和组织使用网络应当遵守宪法法律，遵守公共秩序，尊重社会公德，不得利用网络从事危害国家安全、荣誉和利益，未经授权请勿利用文章中的技术资料对任何计算机系统进行入侵操作。利用…
阅读更多...
demo(三)eurekaribbonhystrix----服务降级熔断

demo(三)eurekaribbonhystrix----服务降级熔断

一、介绍&#xff1a; 1、雪崩&#xff1a; 多个微服务之间调用的时候&#xff0c;假如微服务A调用微服务B和微服务C&#xff0c;微服务B和微服务C又调用其他的微服务&#xff0c;这就是所谓的"扇出"。如果扇出的链路上某个微服务的调用响应的时间过长或者不可用&am…
阅读更多...
S7-1200PLC 作为MODBUSTCP服务器通信(多客户端访问)

S7-1200PLC 作为MODBUSTCP服务器通信(多客户端访问)

S7-1200PLC作为MODBUSTCP服务器端通信编程应用，详细内容请查看下面文章链接： ModbusTcp通信(S7-1200PLC作为服务器端)-CSDN博客文章浏览阅读239次。S7-200Smart plc作为ModbusTcp服务器端的通信S7-200SMART PLC ModbusTCP通信(ModbusTcp服务器)_s7-200 modbustcp-CSDN博客文…
阅读更多...
flutter开发web应用支持浏览器跨域设置

flutter开发web应用支持浏览器跨域设置

开发web应用难免会遇到跨域问题&#xff0c;所以flutter设置允许web跨域的设置是要在你的flutter安装路径下面 flutter\bin\cache 找到flutter_tools.stamp文件&#xff0c;然后删除掉&#xff1a;这个文件是临时缓存文件 然后找到 flutter\packages\flutter_tools\lib\src\web…
阅读更多...
Linux使用ifconifg命令，没有显示ens33

Linux使用ifconifg命令，没有显示ens33

Linux使用ifconifg命令&#xff0c;没有显示ens33 1.问题2.步骤2.1 查看虚拟机的组件是否启动了2.2 修改网络配置文件 ONBOOT修改为yes2.3 重启网络2.4 修改网络服务配置 3.解决 1.问题 打开虚拟机准备使用xshell连接时发现连接失败&#xff0c;在机器上查看ip发现ens33不现实…
阅读更多...
【如何学习Python自动化测试】—— 页面元素定位

【如何学习Python自动化测试】—— 页面元素定位

接上篇自动化测试环境搭建&#xff0c;现在我们介绍 webdriver 对浏览器操作的 API。 2、 页面元素定位 通过自动化操作 web 页面&#xff0c;首先要解决的问题就是定位到要操作的对象&#xff0c;比如要模拟用户在页面上的输入框中输入一段字符串&#xff0c;那就必须得定位到…
阅读更多...
【SQL server】 表结构的约束和维护

【SQL server】 表结构的约束和维护

表结构的约束和维护 修改表结构 (1)添加列 (2)删除列 (3)修改列alter table 表名 add 新列名 数据类型给员工表添加一列邮箱 alter table People add PeopleMail varchar(200)删除列 alter table People drop column PeopleMain修改列 alter table 表名 alter column 列名 数据…
阅读更多...
Sa-Token 整合Java17和SpringBoot

Sa-Token 整合Java17和SpringBoot

目录 前言引入项目开启登录认证路由拦截鉴权解决兼容问题总结 前言 之前无意中发现Sa-Token权限认证框架&#xff0c;项目十分好用。 项目地址&#xff1a; https://github.com/dromara/sa-token 官网地址&#xff1a; https://sa-token.cc/doc.html#/start/example 我的个人…
阅读更多...
助力水泥基建裂痕自动化巡检，基于yolov5融合ASPP开发构建多尺度融合目标检测识别系统

助力水泥基建裂痕自动化巡检，基于yolov5融合ASPP开发构建多尺度融合目标检测识别系统

道路场景下的自动化智能巡检、洞体场景下的壁体类建筑缺陷自动检测识别等等已经在现实生活中不断地落地应用了&#xff0c;在我们之前的很多博文中也已经有过很多相关的实践项目经历了&#xff0c;本文的核心目的是想要融合多尺度感受野技术到yolov5模型中以期在较低参数量的情…
阅读更多...
迭代新品 | 第四代可燃气体监测仪，守护燃气管网安全快人一步

迭代新品 | 第四代可燃气体监测仪，守护燃气管网安全快人一步

城市地下市政基础设施是城市有序运行的生命线&#xff0c;事关城市安全、健康运行和高质量发展。近年来&#xff0c;我国燃气事故多发、频发。2020、2021、2022 年分别发生燃气事故668、1140 起、802 起&#xff0c;造成92、106、66 人死亡&#xff0c;560、763、487 人受伤。尤…
阅读更多...
【数据结构】图的存储结构及实现（邻接表和十字链表）

【数据结构】图的存储结构及实现（邻接表和十字链表）

一.邻接矩阵的空间复杂度 假设图G有n个顶点e条边&#xff0c;则存储该图需要O&#xff08;n^2) 不适用稀疏图的存储 二.邻接表 1.邻接表的存储思想&#xff1a; 对于图的每个顶点vi&#xff0c;将所有邻接于vi的顶点链成一个单链表&#xff0c;称为顶点vi的边表&#xff08…
阅读更多...
自动驾驶-BEV感知综述

自动驾驶-BEV感知综述

BEV感知综述 随着自动驾驶传感器配置多模态化、多源化&#xff0c;将多源信息在unified View下表达变得更加关键。BEV视角下构建的local map对于多源信息融合及理解更加直观简洁&#xff0c;同时对于后续规划控制模块任务的开展也更为方便。BEV感知的核心问题是&#xff1a; …
阅读更多...
asp.net学生成绩评估系统VS开发sqlserver数据库web结构c#编程计算机网页项目

asp.net学生成绩评估系统VS开发sqlserver数据库web结构c#编程计算机网页项目

一、源码特点 asp.net 学生成绩评估系统 是一套完善的web设计管理系统&#xff0c;系统具有完整的源代码和数据库&#xff0c;系统主要采用B/S模式开发。 系统运行视频连接&#xff1a;https://www.bilibili.com/video/BV1Wz4y1A7CG/ 二、功能介绍 本系统使用Microsof…
阅读更多...
基于STM32的外部中断（EXTI）在嵌入式系统中的应用

基于STM32的外部中断（EXTI）在嵌入式系统中的应用

外部中断&#xff08;External Interrupt&#xff0c;EXTI&#xff09;是STM32嵌入式系统中常见且重要的功能之一。它允许外部事件&#xff08;例如按键按下、传感器触发等&#xff09;通过适当的引脚触发中断&#xff0c;从而应用于各种嵌入式系统中。在STM32微控制器中&#…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023