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

(二十八)大数据实战——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 常驻

本文地址&#xff1a;blog.lucien.ink/archives/542 发现每次执行 nvidia-smi 都特别慢&#xff0c;发现是需要 nvidia-persistenced 常驻才可以&#xff0c;这个并不会在安装完驱动之后自动配置&#xff0c;需要手动设置一个自启。 cat <<EOF >> /etc/systemd/sy…...

leetcode 42, 58, 14(*)

42. Trapping Rain Water 1.暴力解法&#xff08;未通过&#xff09; 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配置中心 到此部分即微服务的入门。 总的来说&#xff0c;数据存放的节点数越多&#xff0c;分区容忍性就越高&#xff0c;但要复制更新的次数就越多&#xff0c;一致性就越难保证。同时为了保证一致性&#xff0c;更新所有节点数据所需要的时间就…...

K8S:Yaml文件详解

目录 一.Yaml文件详解 1.Yaml文件格式 2.YAML 语法格式 二.Yaml文件编写及相关概念 1.查看 api 资源版本标签 2.yaml编写案例 &#xff08;2&#xff09;Deployment类型编写nginx服务 (3&#xff09;k8s集群中的port介绍 &#xff08;5&#xff09;快速编写yaml文件 …...

机器人连续位姿同步插值轨迹规划—对数四元数、b样条曲线、c2连续位姿同步规划

简介&#xff1a;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格式轻量化压缩的遇到常见问题与处理方法分析 三维模型的轻量化压缩是一项技术挑战&#xff0c;特别是在处理复杂的3DTile格式时。下面列举了一些处理过程中可能遇到的常见问题以及相应的处理方法&#xff1a; 模型精度损失&#xff1a;在进行压缩处理时&#x…...

2023-简单点-开启防火墙后,ping显示请求超时;windows共享盘挂在不上

情景描述 树莓派 挂载 windows共享盘 之前一直可以&#xff0c;突然有一天不行了 ping xxxx不通了 一查&#xff0c;或许是服务器被同事开了防火墙&#xff0c;默认关闭了ping的回显 操作&#xff1a; 开启ping回显cmd ping通了&#xff0c;但是挂载还是不行, 显示 dmesg命…...

华为Java工程师面试题

常见问题&#xff1a; 什么是Java虚拟机&#xff08;JVM&#xff09;&#xff1f;它与现实中的计算机有什么不同&#xff1f;Java中的基本数据类型有哪些&#xff1f;它们的范围是什么&#xff1f;什么是引用类型&#xff1f;Java中的引用类型有哪些&#xff1f;什么是对象&am…...

大数据Flink(七十四):SQL的滑动窗口(HOP)

文章目录 SQL的滑动窗口(HOP) SQL的滑动窗口(HOP) 滑动窗口定义:滑动窗口也是将元素指定给固定长度的窗口。与滚动窗口功能一样,也有窗口大小的概念。不一样的地方在于,滑动窗口有另一个参数控制窗口计算的频率(滑动窗口滑动的步长)。因此,如果滑动的步长小于窗口大…...

Hystrix和Sentinel熔断降级设计理念

目录 1 基本介绍2 Hystrix信号量和线程池区别2.1 信号量模式2.2 线程池模式2.3 注意 3 Sentinel介绍 1 基本介绍 Sentinel 和 Hystrix 的原则是一致的: 当检测到调用链路中某个资源出现不稳定的表现&#xff0c;例如请求响应时间长或异常比例升高的时候&#xff0c;则对这个资源…...

模型参数、模型存储精度、参数与显存

模型参数量衡量单位 M&#xff1a;百万&#xff08;Million&#xff09; B&#xff1a;十亿&#xff08;Billion&#xff09; 1 B 1000 M 1B 1000M 1B1000M 参数存储精度 模型参数是固定的&#xff0c;但是一个参数所表示多少字节不一定&#xff0c;需要看这个参数以什么…...

Mybatis逆向工程,动态创建实体类、条件扩展类、Mapper接口、Mapper.xml映射文件

今天呢&#xff0c;博主的学习进度也是步入了Java Mybatis 框架&#xff0c;目前正在逐步杨帆旗航。 那么接下来就给大家出一期有关 Mybatis 逆向工程的教学&#xff0c;希望能对大家有所帮助&#xff0c;也特别欢迎大家指点不足之处&#xff0c;小生很乐意接受正确的建议&…...

Python爬虫(二):爬虫完整流程

爬虫完整流程详解&#xff08;7大核心步骤实战技巧&#xff09; 一、爬虫完整工作流程 以下是爬虫开发的完整流程&#xff0c;我将结合具体技术点和实战经验展开说明&#xff1a; 1. 目标分析与前期准备 网站技术分析&#xff1a; 使用浏览器开发者工具&#xff08;F12&…...

Java线上CPU飙高问题排查全指南

一、引言 在Java应用的线上运行环境中&#xff0c;CPU飙高是一个常见且棘手的性能问题。当系统出现CPU飙高时&#xff0c;通常会导致应用响应缓慢&#xff0c;甚至服务不可用&#xff0c;严重影响用户体验和业务运行。因此&#xff0c;掌握一套科学有效的CPU飙高问题排查方法&…...

USB Over IP专用硬件的5个特点

USB over IP技术通过将USB协议数据封装在标准TCP/IP网络数据包中&#xff0c;从根本上改变了USB连接。这允许客户端通过局域网或广域网远程访问和控制物理连接到服务器的USB设备&#xff08;如专用硬件设备&#xff09;&#xff0c;从而消除了直接物理连接的需要。USB over IP的…...

20个超级好用的 CSS 动画库

分享 20 个最佳 CSS 动画库。 它们中的大多数将生成纯 CSS 代码&#xff0c;而不需要任何外部库。 1.Animate.css 一个开箱即用型的跨浏览器动画库&#xff0c;可供你在项目中使用。 2.Magic Animations CSS3 一组简单的动画&#xff0c;可以包含在你的网页或应用项目中。 3.An…...

保姆级【快数学会Android端“动画“】+ 实现补间动画和逐帧动画!!!

目录 补间动画 1.创建资源文件夹 2.设置文件夹类型 3.创建.xml文件 4.样式设计 5.动画设置 6.动画的实现 内容拓展 7.在原基础上继续添加.xml文件 8.xml代码编写 (1)rotate_anim (2)scale_anim (3)translate_anim 9.MainActivity.java代码汇总 10.效果展示 逐帧…...

C++_哈希表

本篇文章是对C学习的哈希表部分的学习分享 相信一定会对你有所帮助~ 那咱们废话不多说&#xff0c;直接开始吧&#xff01; 一、基础概念 1. 哈希核心思想&#xff1a; 哈希函数的作用&#xff1a;通过此函数建立一个Key与存储位置之间的映射关系。理想目标&#xff1a;实现…...

Vue3中的computer和watch

computed的写法 在页面中 <div>{{ calcNumber }}</div>script中 写法1 常用 import { computed, ref } from vue; let price ref(100);const priceAdd () > { //函数方法 price 1price.value ; }//计算属性 let calcNumber computed(() > {return ${p…...

JDK 17 序列化是怎么回事

如何序列化&#xff1f;其实很简单&#xff0c;就是根据每个类型&#xff0c;用工厂类调用。逐个完成。 没什么漂亮的代码&#xff0c;只有有效、稳定的代码。 代码中调用toJson toJson 代码 mapper.writeValueAsString ObjectMapper DefaultSerializerProvider 一堆实…...