当前位置: 首页 > news >正文

Kafka集成flume

1.flume作为生产者集成Kafka

        kafka作为flume的sink,扮演消费者角色

     1.1 flume配置文件

        vim $kafka/jobs/flume-kafka.conf

# agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1 c2# Describe/configure the source
a1.sources.r1.type = TAILDIR
#记录最后监控文件的断点的文件,此文件位置可不改
a1.sources.r1.positionFile =  /export/server/flume/job/data/tail_dir.json
a1.sources.r1.filegroups = f1 f2
a1.sources.r1.filegroups.f1 = /export/server/flume/job/data/.*file.*
a1.sources.r1.filegroups.f2 =/export/server/flume/job/data/.*log.*# Describe the sink
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = customers
a1.sinks.k1.kafka.bootstrap.servers =node1:9092,node2:9092
a1.sinks.k1.kafka.flumeBatchSize = 20
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.linger.ms = 1
a1.sinks.k1.kafka.producer.compression.type = snappy# 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

1.2开启flume监控

flume-ng agent -n a1 -c conf/ -f /export/server/kafka/jobs/kafka-flume.conf

1.3开启Kafka消费者

kafka-console-consumer.sh --bootstrap-server node1:9092,node2:9092 --topic consumers --from-beginning

1.4生产数据

往被监控文件输入数据

[ljr@node1 data]$echo hello >>file2.txt 
[ljr@node1 data]$ echo ============== >>file2.txt 

查看Kafka消费者

可见Kafka集成flume生产者成功。

2.flume作为消费者集成Kafka

        kafka作为flume的source,扮演生产者角色

2.1flume配置文件

vim $kafka/jobs/flume-kafka.conf


# agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1# Describe/configure the source
a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource
#注意不要大于channel transactionCapacity的值100
a1.sources.r1.batchSize = 50 
a1.sources.r1.batchDurationMillis = 200
a1.sources.r1.kafka.bootstrap.servers =node1:9092, node1:9092
a1.sources.r1.kafka.topics = consumers
a1.sources.r1.kafka.consumer.group.id = custom.g.id# 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
#注意transactionCapacity的值不要小于sources batchSize的值50
a1.channels.c1.transactionCapacity = 100# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

2.2开启flume监控

flume-ng agent -n a1 -c conf/ -f /export/server/kafka/jobs/kafka-flume1.conf

2.3开启Kafka生产者并生产数据

kafka-console-producer.sh --bootstrap-server node1:9092,node2:9092 --topic consumers

 

查看flume监控台

可见Kafka集成flume消费者成功。

相关文章:

Kafka集成flume

1.flume作为生产者集成Kafka kafka作为flume的sink,扮演消费者角色 1.1 flume配置文件 vim $kafka/jobs/flume-kafka.conf # agent a1.sources r1 a1.sinks k1 a1.channels c1 c2# Describe/configure the source a1.sources.r1.type TAILDIR #记录最后监控文件…...

如何让视频有高级感 高级感视频制作方法 高级感视频怎么剪 会声会影视频剪辑制作教程 会声会影中文免费下载

高质量视频通常具有清晰的画面、优质的音频和令人印象深刻的视觉效果。这篇文章来了解如何让视频有高级感,高级感视频制作方法。 一、如何让视频有高级感 要让视频有高级感,要注意以下几个要点: 1、剧本和故事性:一个好的剧本和…...

详解|访问学者申请被拒原因有哪些?

访问学者项目为全球科研人员提供了一个难得的机会,让他们能够跨越国界,深入不同的学术环境,进行学术交流和合作。然而,并非所有申请者都能如愿以偿地获得这一机会。本文将对访问学者申请中常见的被拒原因进行详细解析,…...

[鹤城杯 2021]BabyRSA

题目: from Crypto.Util.number import getPrime, bytes_to_long from secret import flagp getPrime(1024) q getPrime(1024) n p * q e 65537 hint1 p >> 724 hint2 q % (2 ** 265) ct pow(bytes_to_long(flag), e, n) print(hint1) print(hint2) p…...

西安市工业倍增引导基金子基金申报条件流程和材料程序指南(2024年)

一、基本情况 产业投资基金是以产业发展为首要目标,围绕经济社会发展规划和产业发展政策,发挥“有效市场”作用,支持重点领域、重点产业、重点区域(如:全市六大支柱产业、五大新兴产业领域成熟期重点规模以上企业以及“…...

微型丝杆的耐用性和延长使用寿命的关键因素!

无论是机械设备,还是精密传动元件,高精度微型丝杆是各种机械设备中不可或缺的重要组件。它的精度和耐用性直接影响着工作效率和产品品质,在工业技术不断进步的情况下,对微型丝杆的性能要求也越来越高,如何提升微型丝杆…...

音频文件下载后,如何轻松转换格式?

在我们日常的数字生活中,下载各种音频文件是司空见惯的事情。然而,有时候我们可能需要将这些音频文件转换为不同的格式,以适应不同的设备或编辑需求。无论您是希望将下载的音频文件转换为通用的MP3格式,还是需要将其转换为高保真的…...

Intel平台,13600KF+3060Ti,虚拟机安装macOS 14(2024年6月)

距离上次装macOS虚拟机已经有一段时间了,macOS系统现在大版本升级的速度也是越来越快了,由于Office只支持最新三个版本的macOS,所以现在保底也得安装macOS 12了,我这次是用macOS 14做实验,13和12的安装方式和macOS 14一…...

Cookie、Session、Token的关系和区别

关系 Session与Cookie:Session通常依赖于Cookie来工作。当服务器为客户端创建一个Session时,它会在服务器上存储与客户端相关的信息,并将一个唯一的SessionID通过Cookie发送给客户端。客户端在后续的请求中会携带这个Cookie(包含…...

Windows 11 中安装 Docker Desktop 并安装镜像

本该主要介绍在 Windows 11 中安装 Docker Desktop 时的一些准备工作,以及该如何下载和安装,然后分别使用管理界面和 Docker 命令安装两个镜像。 一、准备工作 在 Windows 11 中安装 Docker Desktop 前,需要做一些准备。打开 【Windows 功能…...

深入剖析Java线程池之“newWorkStealingPool“

1. 概述 newWorkStealingPool 是Java 8中引入的一个新型线程池,它基于ForkJoinPool实现,并采用了“工作窃取”(Work-Stealing)算法。这种线程池特别适用于可并行化且计算密集型的任务,能够充分利用多核CPU资源,提高任务执行效率。 2. 工作窃取算法(Work-Stealing Algor…...

《跟我一起学“网络安全”》——安全设备

安全设备 一、安全设备–IDS IDS入侵检测 (1)什么是入侵检测: 入侵检测系统(intrusion detection system,简称“IDS”)是一种对网络传输进行即时监视,在发现可疑传输时发出警报或者采取主动反应措施的网络安全设备。…...

猜测Tomcat如何实现WebSocket协议

一、WebSocket协议的实现 (一)WebSocket是官方的协议接口标准。 (二)如果一门编程语言可以网络连接和并发,就能创建一种WebSocket实现。 (三)同一种编程语言,有不同的协议实现版本和框架。 二、Tomcat实现 在Tomcat容器中实现了对应的WebSocket版本&am…...

uniApp @input事件更改输入框值,值改变了但是页面没更新新的值

<uni-easyinputtype"text"trim"all":inputBorder"false"v-model"customFormData.completePercent"input"(val) > completeOnInput(val)"placeholder"请输入" /> function completeOnInput(val) {let num…...

两行css 实现瀑布流

html <ul ><li><a href"" ><img src"05094532gc6w.jpg" alt"111" /><p>传奇</p></a></li><li><a href"" ><img src"05094532gc6w.jpg" alt"111"…...

Centos7.9部署单节点K8S环境

Centos7.9部署单节点K8S环境 通过Centos extras镜像源安装K8S环境&#xff0c;优点是方便快捷&#xff0c;缺点是版本较低&#xff0c;安装后的版本为1.5.2。 1. 准备工作 关闭selinux [rootlocalhost ~]# cat /etc/selinux/config# This file controls the state of SELin…...

【CV】stable diffusion初步理解

来自gpt-4o Stable diffusion 和DALLE的关系 Stable Diffusion 和 DALL-E 都是生成图像的人工智能模型&#xff0c;但它们有不同的开发背景和技术实现。 Stable Diffusion: 开发者: 由Stability AI开发&#xff0c;并与CompVis和LAION等组织合作。技术: 基于扩散模型&#xf…...

足底筋膜炎最好的恢复办法

足底筋膜炎是一种由足底筋膜受到炎症刺激而引起的疼痛和不适的疾病。其典型症状主要包括&#xff1a; 1、足底疼痛&#xff1a;这是足底筋膜炎最常见的症状。疼痛通常位于足跟部位&#xff0c;患者可能感到刺痛或灼热感。尤其在早晨起床或长时间站立后&#xff0c;这种疼痛感会…...

Fiddler抓包工具介绍

下载 下载:Web Debugging Proxy and Troubleshooting Tools|Fiddler 进去要填一个表 汉化版 百度网盘 请输入提取码 提取码&#xff1a;xq9t 下载过附件之后分别把两个文件 点开fiddler就ok了 配置https fiddler要想抓到https包(解密的),点击tools->options勾选三个对…...

知乎号开始运营了,宣传一波

知乎号开始发布一些小说、散文还有诗歌了&#xff0c;欢迎大家多来关注 知乎链接&#xff1a;姜亚轲 每篇小说都改编成网易云音乐&#xff0c;文章中也有链接&#xff0c;我做的词&#xff0c;Suno编曲和演唱&#xff0c;欢迎大家来听听...

【Linux】shell脚本忽略错误继续执行

在 shell 脚本中&#xff0c;可以使用 set -e 命令来设置脚本在遇到错误时退出执行。如果你希望脚本忽略错误并继续执行&#xff0c;可以在脚本开头添加 set e 命令来取消该设置。 举例1 #!/bin/bash# 取消 set -e 的设置 set e# 执行命令&#xff0c;并忽略错误 rm somefile…...

中南大学无人机智能体的全面评估!BEDI:用于评估无人机上具身智能体的综合性基准测试

作者&#xff1a;Mingning Guo, Mengwei Wu, Jiarun He, Shaoxian Li, Haifeng Li, Chao Tao单位&#xff1a;中南大学地球科学与信息物理学院论文标题&#xff1a;BEDI: A Comprehensive Benchmark for Evaluating Embodied Agents on UAVs论文链接&#xff1a;https://arxiv.…...

【网络安全产品大调研系列】2. 体验漏洞扫描

前言 2023 年漏洞扫描服务市场规模预计为 3.06&#xff08;十亿美元&#xff09;。漏洞扫描服务市场行业预计将从 2024 年的 3.48&#xff08;十亿美元&#xff09;增长到 2032 年的 9.54&#xff08;十亿美元&#xff09;。预测期内漏洞扫描服务市场 CAGR&#xff08;增长率&…...

为什么需要建设工程项目管理?工程项目管理有哪些亮点功能?

在建筑行业&#xff0c;项目管理的重要性不言而喻。随着工程规模的扩大、技术复杂度的提升&#xff0c;传统的管理模式已经难以满足现代工程的需求。过去&#xff0c;许多企业依赖手工记录、口头沟通和分散的信息管理&#xff0c;导致效率低下、成本失控、风险频发。例如&#…...

Objective-C常用命名规范总结

【OC】常用命名规范总结 文章目录 【OC】常用命名规范总结1.类名&#xff08;Class Name)2.协议名&#xff08;Protocol Name)3.方法名&#xff08;Method Name)4.属性名&#xff08;Property Name&#xff09;5.局部变量/实例变量&#xff08;Local / Instance Variables&…...

系统设计 --- MongoDB亿级数据查询优化策略

系统设计 --- MongoDB亿级数据查询分表策略 背景Solution --- 分表 背景 使用audit log实现Audi Trail功能 Audit Trail范围: 六个月数据量: 每秒5-7条audi log&#xff0c;共计7千万 – 1亿条数据需要实现全文检索按照时间倒序因为license问题&#xff0c;不能使用ELK只能使用…...

React Native在HarmonyOS 5.0阅读类应用开发中的实践

一、技术选型背景 随着HarmonyOS 5.0对Web兼容层的增强&#xff0c;React Native作为跨平台框架可通过重新编译ArkTS组件实现85%以上的代码复用率。阅读类应用具有UI复杂度低、数据流清晰的特点。 二、核心实现方案 1. 环境配置 &#xff08;1&#xff09;使用React Native…...

oracle与MySQL数据库之间数据同步的技术要点

Oracle与MySQL数据库之间的数据同步是一个涉及多个技术要点的复杂任务。由于Oracle和MySQL的架构差异&#xff0c;它们的数据同步要求既要保持数据的准确性和一致性&#xff0c;又要处理好性能问题。以下是一些主要的技术要点&#xff1a; 数据结构差异 数据类型差异&#xff…...

Java多线程实现之Callable接口深度解析

Java多线程实现之Callable接口深度解析 一、Callable接口概述1.1 接口定义1.2 与Runnable接口的对比1.3 Future接口与FutureTask类 二、Callable接口的基本使用方法2.1 传统方式实现Callable接口2.2 使用Lambda表达式简化Callable实现2.3 使用FutureTask类执行Callable任务 三、…...

屋顶变身“发电站” ,中天合创屋面分布式光伏发电项目顺利并网!

5月28日&#xff0c;中天合创屋面分布式光伏发电项目顺利并网发电&#xff0c;该项目位于内蒙古自治区鄂尔多斯市乌审旗&#xff0c;项目利用中天合创聚乙烯、聚丙烯仓库屋面作为场地建设光伏电站&#xff0c;总装机容量为9.96MWp。 项目投运后&#xff0c;每年可节约标煤3670…...