(二十八)大数据实战——Flume数据采集之kafka数据生产与消费集成案例
前言
本节内容我们主要介绍一下flume数据采集和kafka消息中间键的整合。通过flume监听nc端口的数据,将数据发送到kafka消息的first主题中,然后在通过flume消费kafka中的主题消息,将消费到的消息打印到控制台上。集成使用flume作为kafka的生产者和消费者。关于nc工具、flume以及kafka的安装部署,这里不在赘述,请读者查看作者往期博客内容。整体架构如下:

正文
-
启动Kafka集群,创建first主题
- 启动Kafka集群
- 创建first主题
kafka-topics.sh --bootstrap-server hadoop101:9092 --create --topic first --partitions 3 --replication-factor 3
- 查看first主题详情
kafka-topics.sh --bootstrap-server hadoop101:9092 --describe --topic first
- 在hadoop101服务器flume安装目录/opt/module/apache-flume-1.9.0/job下创建nc监听服务
- 创建nc监听的flume任务:job-netcat-flume-kafka.conf
# 1 组件定义 a1.sources = r1 a1.sinks = k1 a1.channels = c1 # 2 配置 source a1.sources.r1.type = netcat a1.sources.r1.bind = hadoop101 a1.sources.r1.port = 1111 # 3 配置 channel a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100 # 4 配置 sink a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink a1.sinks.k1.kafka.bootstrap.servers = hadoop101:9092,hadoop102:9092,hadoop103:9092 a1.sinks.k1.kafka.topic = first a1.sinks.k1.kafka.flumeBatchSize = 20 a1.sinks.k1.kafka.producer.acks = 1 a1.sinks.k1.kafka.producer.linger.ms = 1 # 5 拼接组件 a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1
- 在hadoop102服务器flume安装目录/opt/module/apache-flume-1.9.0/job下创建kafka监听r任务
- 创建kafka监听的flume任务:job-kafka-flume-console.conf
# 1 组件定义 a1.sources = r1 a1.sinks = k1 a1.channels = c1 # 2 配置 source a1.sources.r1.type = org.apache.flume.source.kafka.KafkaSource a1.sources.r1.batchSize = 50 a1.sources.r1.batchDurationMillis = 200 a1.sources.r1.kafka.bootstrap.servers = hadoop101:9092,hadoop102:9092,hadoop103:9092 a1.sources.r1.kafka.topics = first a1.sources.r1.kafka.consumer.group.id = custom.g.id # 3 配置 channel a1.channels.c1.type = memory a1.channels.c1.capacity = 1000 a1.channels.c1.transactionCapacity = 100 # 4 配置 sink a1.sinks.k1.type = logger # 5 拼接组件 a1.sources.r1.channels = c1 a1.sinks.k1.channel = c1
- 在hadoop102服务器启动kafka监听任务job-kafka-flume-console.conf
- 启动job-kafka-flume-console.conf任务
bin/flume-ng agent -c conf/ -n a1 -f job/job-kafka-flume-console.conf -Dflume.root.logger=INFO,console
- 在hadoop101服务器启动nc监听任务job-netcat-flume-kafka.conf
- 启动job-netcat-flume-kafka.conf任务
bin/flume-ng agent -c conf/ -n a1 -f job/job-netcat-flume-kafka.conf -Dflume.root.logger=INFO,console
- 使用netcat工具发送数据到nc服务1111端口
- 发送nc消息
- 查看结果
- 控制台结果
结语
该案例证明了flume1成功采集到了nc监听端口的数据,并将数据发送到了kafka主题first中,flume2成功从kafka主题中消费到了数据并打印到了控制台。关于Flume数据采集之kafka数据生产与消费的集成案例到这里就结束了,我们下期见。。。。。。
相关文章:
(二十八)大数据实战——Flume数据采集之kafka数据生产与消费集成案例
前言 本节内容我们主要介绍一下flume数据采集和kafka消息中间键的整合。通过flume监听nc端口的数据,将数据发送到kafka消息的first主题中,然后在通过flume消费kafka中的主题消息,将消费到的消息打印到控制台上。集成使用flume作为kafka的生产…...
vue3:22、vue-router的使用
import { createRouter, createWebHistory } from vue-router//history模式:createWebHistory //hash模式:createWebHashHistory//vite中的环境变量 import.meta.env.BASE_URL 就是vite.config.js中的base配置项 const router createRouter({history:…...
深入理解JVM虚拟机第五篇:一些常用的JVM虚拟机(二)
文章目录 一:JRockit VM的介绍 二:J9 VM的介绍 三:KVM和CDC/CLDC Hotspot 四:Azul VM的介绍 五:Liquid VM的介绍 六:Apache Harmoney 七:Microsoft JVM 八:Taobao JVM 九&a…...
导数公式及求导法则
目录 基本初等函数的导数公式 求导法则 有理运算法则 复合函数求导法 隐函数求导法 反函数求导法 参数方程求导法 对数求导法 基本初等函数的导数公式 基本初等函数的导数公式包括: C0(x^n)nx^(n-1)(a^x)a^x*lna(e^x)e^x(loga(x))1/(xlna)(lnx)1/x(sinx)cos…...
SpringMVC系列(六)之JSON数据返回以及异常处理机制
目录 前言 一. JSON概述 二. JSON数据返回 1. 导入pom依赖 2. 添加配置文件(spring-mvc.xml) 3. ResponseBody注解使用 4. 效果展示 5. Jackson介绍 三. 全局异常处理 1. 为什么要全局异常处理 2. 异常处理思路 3. 异常处理方式一 4. 异常处…...
民安智库(北京第三方窗口测评)开展汽车消费者焦点小组座谈会调查
民安智库近日开展了一场汽车消费者焦点小组座谈会,旨在深入了解目标消费者对汽车功能的需求和消费习惯,为汽车企业提供有针对性的解决方案。 在焦点小组座谈会中,民安智库公司(第三方市容环境指数测评)邀请了一群具有…...
【CVPR2021】MVDNet论文阅读分析与总结
Challenge: 现有的目标检测器主要融合激光雷达和相机,通常提供丰富和冗余的视觉信息 利用最先进的成像雷达,其分辨率比RadarNet和LiRaNet中使用的分辨率要细得多,提出了一种有效的深度后期融合方法来结合雷达和激光雷达信号。 MV…...
IDEA指定Maven settings file文件未生效
背景:在自己电脑上配置的时候,由于公司项目和我自己的项目的Maven仓库不一致,我就在项目中指定了各自的Maven配置文件。但是我发现公司的项目私有仓库地址IDEA总是识别不到! 俩个配置文件分别是: /Users/sml/Mine/研发…...
swift UI 和UIKIT 如何配合使用
SwiftUI和UIKit可以在同一个iOS应用程序中配合使用。它们是两个不同的用户界面框架,各自有自己的优势和特点。在现实开发中,很多iOS应用程序并不是一开始就完全采用SwiftUI或UIKit,而是根据需要逐步引入SwiftUI或者使用两者共存。 SwiftUI的…...
c语言练习题55:IP 地址⽆效化
IP 地址⽆效化 题⽬描述: 给你⼀个有效的 IPv4 地址 address ,返回这个 IP 地址的⽆效化版本。 所谓⽆效化 IP 地址,其实就是⽤ "[.]" 代替了每个 "."。 • ⽰例 1: 输⼊:address "1.1.1.…...
nvidia-persistenced 常驻
本文地址:blog.lucien.ink/archives/542 发现每次执行 nvidia-smi 都特别慢,发现是需要 nvidia-persistenced 常驻才可以,这个并不会在安装完驱动之后自动配置,需要手动设置一个自启。 cat <<EOF >> /etc/systemd/sy…...
leetcode 42, 58, 14(*)
42. Trapping Rain Water 1.暴力解法(未通过) class Solution { public:int trap(vector<int>& height) {int n height.size();int res 0;for(int i0; i<n; i){int r_max 0, l_max 0;for(int j i; j<n; j)r_max max(r_max, heigh…...
SpringCloud-微服务CAP原则
接上文 SpringCloud-Config配置中心 到此部分即微服务的入门。 总的来说,数据存放的节点数越多,分区容忍性就越高,但要复制更新的次数就越多,一致性就越难保证。同时为了保证一致性,更新所有节点数据所需要的时间就…...
K8S:Yaml文件详解
目录 一.Yaml文件详解 1.Yaml文件格式 2.YAML 语法格式 二.Yaml文件编写及相关概念 1.查看 api 资源版本标签 2.yaml编写案例 (2)Deployment类型编写nginx服务 (3)k8s集群中的port介绍 (5)快速编写yaml文件 …...
机器人连续位姿同步插值轨迹规划—对数四元数、b样条曲线、c2连续位姿同步规划
简介:Smooth orientation planning is benefificial for the working performance and service life of industrial robots, keeping robots from violent impacts and shocks caused by discontinuous orientation planning. Nevertheless, the popular used quate…...
三维模型3DTile格式轻量化压缩的遇到常见问题与处理方法分析
三维模型3DTile格式轻量化压缩的遇到常见问题与处理方法分析 三维模型的轻量化压缩是一项技术挑战,特别是在处理复杂的3DTile格式时。下面列举了一些处理过程中可能遇到的常见问题以及相应的处理方法: 模型精度损失:在进行压缩处理时&#x…...
2023-简单点-开启防火墙后,ping显示请求超时;windows共享盘挂在不上
情景描述 树莓派 挂载 windows共享盘 之前一直可以,突然有一天不行了 ping xxxx不通了 一查,或许是服务器被同事开了防火墙,默认关闭了ping的回显 操作: 开启ping回显cmd ping通了,但是挂载还是不行, 显示 dmesg命…...
华为Java工程师面试题
常见问题: 什么是Java虚拟机(JVM)?它与现实中的计算机有什么不同?Java中的基本数据类型有哪些?它们的范围是什么?什么是引用类型?Java中的引用类型有哪些?什么是对象&am…...
大数据Flink(七十四):SQL的滑动窗口(HOP)
文章目录 SQL的滑动窗口(HOP) SQL的滑动窗口(HOP) 滑动窗口定义:滑动窗口也是将元素指定给固定长度的窗口。与滚动窗口功能一样,也有窗口大小的概念。不一样的地方在于,滑动窗口有另一个参数控制窗口计算的频率(滑动窗口滑动的步长)。因此,如果滑动的步长小于窗口大…...
Hystrix和Sentinel熔断降级设计理念
目录 1 基本介绍2 Hystrix信号量和线程池区别2.1 信号量模式2.2 线程池模式2.3 注意 3 Sentinel介绍 1 基本介绍 Sentinel 和 Hystrix 的原则是一致的: 当检测到调用链路中某个资源出现不稳定的表现,例如请求响应时间长或异常比例升高的时候,则对这个资源…...
浏览器访问 AWS ECS 上部署的 Docker 容器(监听 80 端口)
✅ 一、ECS 服务配置 Dockerfile 确保监听 80 端口 EXPOSE 80 CMD ["nginx", "-g", "daemon off;"]或 EXPOSE 80 CMD ["python3", "-m", "http.server", "80"]任务定义(Task Definition&…...
【Axure高保真原型】引导弹窗
今天和大家中分享引导弹窗的原型模板,载入页面后,会显示引导弹窗,适用于引导用户使用页面,点击完成后,会显示下一个引导弹窗,直至最后一个引导弹窗完成后进入首页。具体效果可以点击下方视频观看或打开下方…...
Qt Widget类解析与代码注释
#include "widget.h" #include "ui_widget.h"Widget::Widget(QWidget *parent): QWidget(parent), ui(new Ui::Widget) {ui->setupUi(this); }Widget::~Widget() {delete ui; }//解释这串代码,写上注释 当然可以!这段代码是 Qt …...
理解 MCP 工作流:使用 Ollama 和 LangChain 构建本地 MCP 客户端
🌟 什么是 MCP? 模型控制协议 (MCP) 是一种创新的协议,旨在无缝连接 AI 模型与应用程序。 MCP 是一个开源协议,它标准化了我们的 LLM 应用程序连接所需工具和数据源并与之协作的方式。 可以把它想象成你的 AI 模型 和想要使用它…...
CMake 从 GitHub 下载第三方库并使用
有时我们希望直接使用 GitHub 上的开源库,而不想手动下载、编译和安装。 可以利用 CMake 提供的 FetchContent 模块来实现自动下载、构建和链接第三方库。 FetchContent 命令官方文档✅ 示例代码 我们将以 fmt 这个流行的格式化库为例,演示如何: 使用 FetchContent 从 GitH…...
Web 架构之 CDN 加速原理与落地实践
文章目录 一、思维导图二、正文内容(一)CDN 基础概念1. 定义2. 组成部分 (二)CDN 加速原理1. 请求路由2. 内容缓存3. 内容更新 (三)CDN 落地实践1. 选择 CDN 服务商2. 配置 CDN3. 集成到 Web 架构 …...
虚拟电厂发展三大趋势:市场化、技术主导、车网互联
市场化:从政策驱动到多元盈利 政策全面赋能 2025年4月,国家发改委、能源局发布《关于加快推进虚拟电厂发展的指导意见》,首次明确虚拟电厂为“独立市场主体”,提出硬性目标:2027年全国调节能力≥2000万千瓦࿰…...
C# 表达式和运算符(求值顺序)
求值顺序 表达式可以由许多嵌套的子表达式构成。子表达式的求值顺序可以使表达式的最终值发生 变化。 例如,已知表达式3*52,依照子表达式的求值顺序,有两种可能的结果,如图9-3所示。 如果乘法先执行,结果是17。如果5…...
CVPR2025重磅突破:AnomalyAny框架实现单样本生成逼真异常数据,破解视觉检测瓶颈!
本文介绍了一种名为AnomalyAny的创新框架,该方法利用Stable Diffusion的强大生成能力,仅需单个正常样本和文本描述,即可生成逼真且多样化的异常样本,有效解决了视觉异常检测中异常样本稀缺的难题,为工业质检、医疗影像…...
区块链技术概述
区块链技术是一种去中心化、分布式账本技术,通过密码学、共识机制和智能合约等核心组件,实现数据不可篡改、透明可追溯的系统。 一、核心技术 1. 去中心化 特点:数据存储在网络中的多个节点(计算机),而非…...








