【大数据之Flume】四、Flume进阶之复制和多路复用、负载均衡和故障转移、聚合案例

1 复制和多路复用

(1)需求:使用 Flume-1 监控文件变动(可以用Exec Source或Taildir Source),Flume-1 将变动内容传递给 Flume-2(用Avro Sink传),(用Avro Source接)Flume-2 负责存储到 HDFS。同时 Flume-1 将变动内容传递给 Flume-3,Flume-3 负责输出到 Local FileSystem。

(2)分析:
在这里插入图片描述
步骤:
(1)在/opt/module/flume-1.9.0/job 目录下创建 group1 文件夹,在/opt/module/flume-1.9.0/目录下创建 data 文件夹,在该文件夹下创建flume文件夹。

(2)在 group1创建 flume-file-flume.conf:配置 1 个接收日志文件的source 和两个 channel、两个 sink,分别输送给 flume-flume- hdfs 和 flume-flume-dir。

vim  flume-file-flume.conf# 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/flume-1.9.0/logs/flume.log
a1.sources.r1.shell = /bin/bash -c# Describethe sink
# sink 端的 avro 是一个数据发送者 
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = hadoop102
a1.sinks.k1.port = 4141# sink 端的 avro 是一个数据发送者
a1.sinks.k2.type= avro
a1.sinks.k2.hostname = hadoop102
a1.sinks.k2.port = 4142# Describe thechannel 
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100a1.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

(3)在group1下创建flume-flume-hdfs.conf:配置上级 Flume 输出的 Source,输出是到 HDFS 的Sink。

vim 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 thechannel 
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

(4)在group1下创建flume-flume-dir.conf:配置上级 Flume 输出的 Source,输出是到本地目录的 Sink。

vim 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 = /opt/module/flume-1.9.0/data/flume# 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

提示:输出的本地目录必须是已经存在的目录,如果该目录不存在,并不会在本地创建新的目录。

(5)先开启HDFS,再分别启动flume-flume-hdfs、flume-flume-dir、flume-file-flume。
  服务器要先开启,再开启客户端,

myhadoop.sh startbin/flume-ng agent -n a2 -c conf/ -f job/group1/flume-flume-hdfs.conf
bin/flume-ng agent -n a3 -c conf/ -f job/group1/flume-flume-dir.conf
bin/flume-ng agent -n a1 -c conf/ -f job/group1/flume-file-flume.conf

(6)检查HDFS上的数据
在这里插入图片描述
在这里插入图片描述

2 负载均衡和故障转移

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

(2)分析:
在这里插入图片描述
步骤:
(1)在/opt/module/flume-1.9.0/job 目录下创建 group2 文件夹,在该文件夹下创建flume-netcat-flume.conf、flume-flume-console1.conf、flume-flume-console2.conf。

(2)flume-netcat-flume.conf:配置 1 个 netcat source 和 1 个 channel、1 个 sink group(2 个 sink),分别输送给 flume-flume-console1.conf、flume-flume-console2.conf。

# Name the components on this agent 
a1.sources = r1
a1.channels = c1
a1.sinkgroups = g1
a1.sinks = k1 k2# Describe/configure the source
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port =44444a1.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# Describe the sink
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = hadoop102
a1.sinks.k1.port = 4141a1.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

(3)创建 flume-flume-console1.conf:配置上级 Flume 输出的 Source,输出是到本地控制台。

# 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

(4)创建 flume-flume-console2.conf:配置上级 Flume 输出的 Source,输出是到本地控制台。

# 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 = hadoop102
a3.sources.r1.port = 4142# 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)执行配置文件,分别开启对应配置文件: flume-flume-console2.conf , flume-flume-console1.conf , flume-netcat-flume.conf。
  同样是需要先开启服务端,再开客户端。

bin/flume-ng agent -c conf/ -n a3 -f job/group2/flume-flume-console2.conf -Dflume.root.logger=INFO,console
bin/flume-ng agent -c conf/ -n a2 -f job/group2/flume-flume-console1.conf -Dflume.root.logger=INFO,console
bin/flume-ng agent -c conf/ -n a1 -f job/group2/flume-netcat-flume.conf

(6)使用 netcat 工具向本机的 44444 端口发送内容

nc localhost 44444

(7)查看Flume2 及 Flume3 的控制台打印日志。
Flume3优先级更高。
在这里插入图片描述
(8)将 Flume3 kill,观察 Flume2 的控制台打印情况。
在这里插入图片描述
在这里插入图片描述
负载均衡需求:
  使用 Flume1 监控一个端口,其 sink 组中的 sink 分别对接 Flume2 和 Flume3,采用FailoverSinkProcessor,实现负载均衡的功能。

步骤:
  只需要修改flume-netcat-flume.conf中a1.sinkgroups.g1.processor的配置内容,把原来相关的内容都删除掉,添加以下,其余相同。

a1.sinkgroups.g1.processor.type = load_balance;
#使用退避算法轮询sink组
a1.sinkgroups.g1.processor.backoff = true;

3 聚合

(1)需求:hadoop102 上的Flume-1 监控文件/opt/module/flume-1.9.0/group.log,hadoop103 上的Flume-2 监控某一个端口的数据流,Flume-1 与 Flume-2 将数据发送给 hadoop104 上的 Flume-3,Flume-3 将最终数据打印到控制台。

(2)分析:
在这里插入图片描述
步骤:
(1)在opt/module/flume-1.9.0/job 目录下创建一个group3 文件夹;分发整个Flume给hadoop103、hadoop104。

(2)在 hadoop102 上创建配置文件flume1-logger-flume.conf:配置 Source 用于监控 hive.log 文件,配置 Sink 输出数据到下一级Flume,并在/opt/module/flume-1.9.0/data下创建空白的文件group.log。

# 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 /opt/module/flume-1.9.0/data/group.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 = hadoop103
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 = 10# Bind the source and sink to the channel
a2.sources.r1.channels = c1
a2.sinks.k1.channel = c1

(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# 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)在各主机上分别执行配置文件:flume3-flume-logger.conf,flume2-netcat-flume.conf,flume1-logger-flume.conf。

[lyx@hadoop104 flume-1.9.0]$ bin/flume-ng agent -c conf/ -n a3 -f job/group3/flume3-flume-logger.conf -Dflume.root.logger=INFO,console
[lyx@hadoop103 flume-1.9.0]$ bin/flume-ng agent -c conf/ -n a2 -f job/group3/flume2-netcat-flume.conf
[lyx@hadoop102 flume-1.9.0]$ bin/flume-ng agent -c conf/ -n a1 -f job/group3/flume1-logger-flume.conf

(6)在 hadoop102 上向/opt/module/flume-1.9.0/data 目录下的 group.log 追加内容

[lyx@hadoop102 data]$ echo 'hello' > group.log

(7)在 hadoop103 上向 44444 端口发送数据

nc hadoop103 44444

(8)查看hadoop104上的数据。
在这里插入图片描述

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

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

相关文章

BugKu CTF(杂项篇MISC)—想要种子吗

BugKu CTF(杂项篇MISC)—想要种子吗 提 示: 描 述:flag{} 题目下载后是一张图片,打开如下。 一、工具 十六进制编辑器010 editor kali系统文件分离工具binwalk或者foremost 维吉尼亚密码 STEGHIDE图片隐写工具 文章所需的软件下载地址 ARCHPR压缩包密码破解…

美团2024校招6000人;伯克利博士讲Llama 2技术细节;互联网转行AIGC最全指北;技术进步周期与创客崛起 | ShowMeAI日报

👀日报&周刊合集 | 🎡生产力工具与行业应用大全 | 🧡 点赞关注评论拜托啦! 🤖 美团 2024 届校园招聘将录用 6000 人,技术类岗位扩招超 50% 美团招聘公众号宣布启动 2024 届校园招聘!此次招聘…

PCB制版技术

1、在头脑里形成一个原理图----现在就下载AD9盖版,诞生了一个问题,电路板去哪里买,买了怎么焊接电路和芯片,怎样流程化批量制作电子产品 1.1 形成一个PCB板,形成一个结构 1.2 焊接,嫁接,组装等 …

Scaling Instruction-Finetuned Language Models

Paper name Scaling Instruction-Finetuned Language Models Paper Reading Note Paper URL: https://arxiv.org/pdf/2210.11416.pdf TL;DR 2022 年谷歌出的文章,对指令微调的影响因素进行分析,提出了一些提升指令微调效果的方案。与该文章一起出品…

vue sku商品规格多选

vue sku商品规格多选 1.创建一个数据对象,用于存储SKU的选中状态。例如,可以使用一个数组来表示选中的SKU,每个元素代表一个SKU选项的id。 data() {return {selectedOptions: []} }2.在SKU选项列表中,使用v-bind:class绑定一个计…

基于双层优化的微电网系统规划设计方法(Matlab代码实现)

目录 💥1 概述 1.1 微电网系统结构 1.2 微电网系统双层规划设计结构 1.3 双层优化模型 1.4 上层容量优化模型 1.5 下层调度优化模型 📚2 运行结果 🎉3 文献来源 🌈4 Matlab代码、数据、文章讲解 💥1 概述 文献来源&…

Meta “地平线世界”移动端应用即将上线,手机快乐元宇宙?

根据海外记者 Janko Roettgers 的报道,Meta 预计很快推出移动版的 VR 元宇宙服务 "地平线世界",这是Meta 长期开发的产品。 根据最新报道,Meta宣布正在研发“地平线世界”的移动版,并表示这一服务已经可以在Quest VR设…

LInux的安装(VMware,网卡设置,SSH连接工具)

Linux的安装 1、安装方式介绍 1.安装方式: 物理机安装:直接将操作系统安装到服务器硬件上 虚拟机安装:通过虚拟机软件安装 **虚拟机( Virtual Machine)**指通过软件模拟的具有完整硬件系统功能、运行在完全隔离环境中的完整计算机系统。 2、安装Linux 在官网将…

PoseiSwap:首个基于模块化设施构建的订单簿 DEX

在前不久,PoseiSwap 曾以1000万美元的估值,获得了来自于ZebecLabs基金会的150万美元的融资。此后 PoseiSwap 又以2500万美元的估值,从GateLabs、EmurgoVentures、Republic以及CipholioVentures等行业顶级投资机构中,获得了新一轮未…

Vue2面试题

1. Vue 的基本原理 当 一 个 Vue 实 例 创 建 时 , Vue 会 遍 历 data 中 的 属 性 , 用 Object.defineProperty ( vue3.0 使 用 proxy) 将 它 们 转 为 getter/setter,并且在内部追踪相关依赖,在属性被访…

Java课题笔记~Maven基础知识

一、什么是Maven? Maven是专门用于管理和构建Java项目的工具。 它的主要功能有: 提供了一套标准化的项目结构提供了一套标准化的构建流程(编译,测试,打包,发布……)提供了一套依赖管理机制 …

子域名收集工具OneForAll的安装与使用-Win

子域名收集工具OneForAll的安装与使用-Win OneForAll是一款功能强大的子域名收集工具 GitHub地址:https://github.com/shmilylty/OneForAll Gitee地址:https://gitee.com/shmilylty/OneForAll 安装 1、python环境准备 OneForAll基于Python 3.6.0开发和…

Kotlin基础(九):对象和委托

前言 本文主要讲解kotlin对象和委托。 Kotlin文章列表 Kotlin文章列表: 点击此处跳转查看 目录 1.1 对象 在Kotlin中,对象(Object)是一个具有特殊用途的单例实例。它是一种创建单个实例的方式,确保在整个应用程序中只存在一个特…

(树) 剑指 Offer 32 - II. 从上到下打印二叉树 II ——【Leetcode每日一题】

❓剑指 Offer 32 - II. 从上到下打印二叉树 II 难度:简单 从上到下按层打印二叉树,同一层的节点按从左到右的顺序打印,每一层打印到一行。 例如: 给定二叉树: [3,9,20,null,null,15,7], 3/ \9 20/ \15 7返回其层次遍历结果&#xff1a…

详解顺序表功能

前言 随着我们C语言的不断深入学习,我们要开始学习一点数据结构来增加我们的内功了,虽说现在很多高级语言的顺序表,链表等可以不用自己实现,但在C语言中是需要我们自己来实现的,这并不能说明C语言和其他语言比C语言很…

使用WebMvcConfigurationSupport后导致原来返回的json数据变为了xml的解决方法

问题 未使用WebMvcConfigurationSupport拦截时返回的数据都是JSON格式,使用WebMvcConfigurationSupport做拦截后数据的返回变为了XML的格式。 原因 在Spring框架中,WebMvcConfigurationSupport 是一个类,它可以用于自定义Spring MVC的配置…

【模型预测控制MPC】使用离散、连续、线性或非线性模型对预测控制进行建模(Matlab代码实现)

💥💥💞💞欢迎来到本博客❤️❤️💥💥 🏆博主优势:🌞🌞🌞博客内容尽量做到思维缜密,逻辑清晰,为了方便读者。 ⛳️座右铭&a…

LLM微调 | Adapter: Parameter-Efficient Transfer Learning for NLP

目的:大模型预训练+微调范式,微调成本高。adapter只只微调新增的小部分参数【但adapter增加了模型层数,引入了额外的推理延迟。】 Adapters最初来源于CV领域的《Learning multiple visual domains with residual adapters》一文,其核心思想是在神经网络模块基础上添加一些残…

【Python机器学习】实验04(1) 多分类(基于逻辑回归)实践

文章目录 多分类以及机器学习实践如何对多个类别进行分类1.1 数据的预处理1.2 训练数据的准备1.3 定义假设函数,代价函数,梯度下降算法(从实验3复制过来)1.4 调用梯度下降算法来学习三个分类模型的参数1.5 利用模型进行预测1.6 评…

CS162 13-17 虚拟内存

起源 为啥我们需要虚拟内存-----------需求是啥? 可以给程序提供一个统一的视图,比如多个程序运行同一个代码段的话,同一个kernel,就可以直接共享 cpu眼里的虚拟内存 无限内存的假象 设计迭代过程 为啥这样设计? 一…