Flume之使用Loadbalancing Sink Processor实现sink负载均衡

2024-04-15 14:58

本文主要是介绍Flume之使用Loadbalancing Sink Processor实现sink负载均衡,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!


前言

  • Load balancing Sink Processor,顾名思义,即能够对Sink组中的每个Sink实现负载均衡,默认采用的是轮询round_robin的方式,还可以使用随机方式random,或者用户自己实现AbstractSinkSelector抽象类定义自己的Sink Selector类,并提供FQCN(Full Qualified Class Name)全类名来进行配置,并且Load balancing Sink Processor还提供了指数退避backoff,即当某个Sink挂掉时,将会将其加入到黑名单,一定时间内不再访问此Sink,退避时间呈指数增长并默认最大值为30000ms,可以手动设置

使用示例

1)flume1.properties

# flume1:此配置用于监控某个端口将其追加内容输出到flume2和flume3中
# 并将两个Sink组成一个sink group,并将Sink Processor设置成load_balance类型
# a1:Netcat Source-> Memory Channel-> Load balancing Sink Processor-> Avro Sink# Agent
a1.sources = r1
a1.channels = c1
a1.sinks = k1 k2# Sink groups
a1.sinkgroups = g1
# 设置sink group中的sinks
a1.sinkgroups.g1.sinks = k1 k2
# 配置Load balancing Sink Processor(只有sink group才可以使用sink processor)
a1.sinkgroups.g1.processor.type = load_balance
# 设置开启指数避让
a1.sinkgroups.g1.processor.backoff = true
# 设置Processor的selector为轮询round_robin
a1.sinkgroups.g1.processor.selector = round_robin
# 设置最大避让时间(ms)
a1.sinkgroups.g1.processor.maxTimeOut = 10000# Sources
# 配置a1.sources.r1的各项属性参数,类型/绑定主机ip/端口号
a1.sources.r1.type = netcat
a1.sources.r1.bind = hadoop101
a1.sources.r1.port = 44444# Channels
# 配置a1.channerls.c1的各项属性参数,缓存方式/最多缓存的Event个数/单次传输的Event个数
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100# Sinks
# sinks.k1
a1.sinks.k1.type = avro
a1.sinks.k1.hostname = hadoop102
a1.sinks.k1.port = 4141
# sinks.k2
a1.sinks.k2.type = avro
a1.sinks.k2.hostname = hadoop103
a1.sinks.k2.port = 4141# Bind
# 注意:source可以绑定多个channel,但是sink/sink group只能绑定单个channel
# r1->c1->g1
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
a1.sinks.k2.channel = c1

2)flume2.properties

# flume2:此配置用于将来自指定Avro端口的数据输出到控制台
# a2:Avro Source->Memory Channel->Logger Sink# Agent
a2.sources = r1
a2.channels = c1
a2.sinks = k1# Sources
# a2.sources.r1
a2.sources.r1.type = avro
# 设置监听本地IP
a2.sources.r1.bind = 0.0.0.0
# 设置监听端口号
a2.sources.r1.port = 4141# Channels
# a2.channels.c1
# 使用内存作为缓存/最多缓存的Event个数/单次传输的Event个数
a2.channels.c1.type = memory
a2.channels.c1.capacity = 1000
a2.channels.c1.transactionCapacity = 100# Sinks
# 运行时设置参数 -Dflume.root.logger=INFO,console 即输出到控制台实时显示
a2.sinks.k1.type = logger
# 设置Event的Body中写入log的最大字节数(默认值为16)
a2.sinks.k1.maxBytesToLog = 256# Bind
a2.sources.r1.channels = c1
a2.sinks.k1.channel = c1

3)flume3.properties

# flume3:此配置用于将来自指定Avro端口的数据输出到控制台
# a3:Avro Source->Memory Channel->Logger Sink# Agent
a3.sources = r1
a3.channels = c1
a3.sinks = k1# Sources
# a3.sources.r1
a3.sources.r1.type = avro
# 设置监听本地IP
a3.sources.r1.bind = 0.0.0.0
# 设置监听端口号
a3.sources.r1.port = 4141# Channels
# a3.channels.c1
# 使用内存作为缓存/最多缓存的Event个数/单次传输的Event个数
a3.channels.c1.type = memory
a3.channels.c1.capacity = 1000
a3.channels.c1.transactionCapacity = 100# Sinks
# 运行时设置参数 -Dflume.root.logger=INFO,console 即输出到控制台实时显示
a3.sinks.k1.type = logger
# 设置Event的Body中写入log的最大字节数(默认值为16)
a3.sinks.k1.maxBytesToLog = 256# Bind
a3.sources.r1.channels = c1
a3.sinks.k1.channel = c1

4)启动命令

Flume Agent a1至a3分别运行在主机hadoop101、hadoop102、hadoop103上

./bin/flume-ng agent -n a1 -c conf -f flume1.properties
./bin/flume-ng agent -n a2 -c conf -f flume2.properties -Dflume.root.logger=INFO,console
./bin/flume-ng agent -n a3 -c conf -f flume3.properties -Dflume.root.logger=INFO,console

5)实现功能

agent a1将指定端口的监听数据采用轮询的方式传输给a2和a3,并分别输出到各自的控制台


End~

这篇关于Flume之使用Loadbalancing Sink Processor实现sink负载均衡的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/906136

相关文章

Java枚举类实现Key-Value映射的多种实现方式

《Java枚举类实现Key-Value映射的多种实现方式》在Java开发中,枚举(Enum)是一种特殊的类,本文将详细介绍Java枚举类实现key-value映射的多种方式,有需要的小伙伴可以根据需要... 目录前言一、基础实现方式1.1 为枚举添加属性和构造方法二、http://www.cppcns.co

使用Python实现快速搭建本地HTTP服务器

《使用Python实现快速搭建本地HTTP服务器》:本文主要介绍如何使用Python快速搭建本地HTTP服务器,轻松实现一键HTTP文件共享,同时结合二维码技术,让访问更简单,感兴趣的小伙伴可以了... 目录1. 概述2. 快速搭建 HTTP 文件共享服务2.1 核心思路2.2 代码实现2.3 代码解读3.

Elasticsearch 在 Java 中的使用教程

《Elasticsearch在Java中的使用教程》Elasticsearch是一个分布式搜索和分析引擎,基于ApacheLucene构建,能够实现实时数据的存储、搜索、和分析,它广泛应用于全文... 目录1. Elasticsearch 简介2. 环境准备2.1 安装 Elasticsearch2.2 J

使用C#代码在PDF文档中添加、删除和替换图片

《使用C#代码在PDF文档中添加、删除和替换图片》在当今数字化文档处理场景中,动态操作PDF文档中的图像已成为企业级应用开发的核心需求之一,本文将介绍如何在.NET平台使用C#代码在PDF文档中添加、... 目录引言用C#添加图片到PDF文档用C#删除PDF文档中的图片用C#替换PDF文档中的图片引言在当

Java中List的contains()方法的使用小结

《Java中List的contains()方法的使用小结》List的contains()方法用于检查列表中是否包含指定的元素,借助equals()方法进行判断,下面就来介绍Java中List的c... 目录详细展开1. 方法签名2. 工作原理3. 使用示例4. 注意事项总结结论:List 的 contain

C#使用SQLite进行大数据量高效处理的代码示例

《C#使用SQLite进行大数据量高效处理的代码示例》在软件开发中,高效处理大数据量是一个常见且具有挑战性的任务,SQLite因其零配置、嵌入式、跨平台的特性,成为许多开发者的首选数据库,本文将深入探... 目录前言准备工作数据实体核心技术批量插入:从乌龟到猎豹的蜕变分页查询:加载百万数据异步处理:拒绝界面

Android中Dialog的使用详解

《Android中Dialog的使用详解》Dialog(对话框)是Android中常用的UI组件,用于临时显示重要信息或获取用户输入,本文给大家介绍Android中Dialog的使用,感兴趣的朋友一起... 目录android中Dialog的使用详解1. 基本Dialog类型1.1 AlertDialog(

MySQL双主搭建+keepalived高可用的实现

《MySQL双主搭建+keepalived高可用的实现》本文主要介绍了MySQL双主搭建+keepalived高可用的实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,... 目录一、测试环境准备二、主从搭建1.创建复制用户2.创建复制关系3.开启复制,确认复制是否成功4.同

Python使用自带的base64库进行base64编码和解码

《Python使用自带的base64库进行base64编码和解码》在Python中,处理数据的编码和解码是数据传输和存储中非常普遍的需求,其中,Base64是一种常用的编码方案,本文我将详细介绍如何使... 目录引言使用python的base64库进行编码和解码编码函数解码函数Base64编码的应用场景注意

Java实现文件图片的预览和下载功能

《Java实现文件图片的预览和下载功能》这篇文章主要为大家详细介绍了如何使用Java实现文件图片的预览和下载功能,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... Java实现文件(图片)的预览和下载 @ApiOperation("访问文件") @GetMapping("