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

Kafka与Flink的整合 -- sink、source

1、首先导入依赖:
        <dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka</artifactId><version>1.15.2</version></dependency>
2、 source:Flink从Kafka中读取数据
public class Demo01KafkaSource {public static void main(String[] args) throws Exception{//构建环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//构建kafka source 环境KafkaSource<String> source = KafkaSource.<String>builder()//指定broker列表.setBootstrapServers("master:9092,node1:9092,node2:9092")//指定topic.setTopics("bigdata")//指定消费组.setGroupId("my-group")//指定数据的读取的位置,earliest指的是读取最早的数据,latest:指定的读取的是最新的数据.setStartingOffsets(OffsetsInitializer.earliest())//读取数据格式:.setValueOnlyDeserializer(new SimpleStringSchema()).build();//使用kafka数据源DataStreamSource<String> kafkaSourceDS = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");kafkaSourceDS.print();//启动flinkenv.execute();}
}
        启动生产kafka:
kafka-console-producer.sh --broker-list master:9092,node1:9092,node2:9092 --topic bigdata
3、sink:Flink向Kafka中写入数据
public class Demo02KafkaSink {public static void main(String[] args) throws Exception{//构建flink的环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//读取数据文件:DataStreamSource<String> studentDS = env.readTextFile("flink/data/students.txt");//创建kafka sinkKafkaSink<String> sink = KafkaSink.<String>builder()//指定flink broker列表.setBootstrapServers("master:9092,node1:9092,node2:9092")//指定数据的格式:.setRecordSerializer(KafkaRecordSerializationSchema.builder()//指定topic,如果topic不存在就会自动的创建一个分区是1个副本是1个的topic.setTopic("student")//指定数据的格式.setValueSerializationSchema(new SimpleStringSchema()).build())//指定数据处理的语义:.setDeliverGuarantee(DeliveryGuarantee.AT_LEAST_ONCE).build();//执行flinkstudentDS.sinkTo(sink);//构建flink环境env.execute();}
}
        启动消费kafka:
kafka-console-consumer.sh --bootstrap-server  master:9092,node1:9092,node2:9092 --from-beginning --topic student

相关文章:

Kafka与Flink的整合 -- sink、source

1、首先导入依赖&#xff1a; <dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka</artifactId><version>1.15.2</version></dependency> 2、 source&#xff1a;Flink从Kafka中读取数据 p…...

小鱼ROS

git clone git clone https://ghproxy.com/https://github.com/stilleshan/ServerStatus git clone 私有仓库 Clone 私有仓库需要用户在 Personal access tokens 申请 Token 配合使用.git clone https://user:your_tokenghproxy.com/https://github.com/your_name/your_priv…...

简单讲讲RISC-V跳转指令基于具体场景的实现

背景 在 RISC-V指令集中&#xff0c;一共有 6 条有条件跳转指令&#xff0c;分别是 beq、bne、blt、bltu、bge、bgeu。如下是它们的定义与接口 BEQ rs1, rs2, imm ≠ BNE rs1, rs2, imm &#xff1c; BLT rs1, rs2, imm ≥ BGE rs1, rs2, imm < unsigned BLTU rs1…...

第13章 Java IO流处理(一) File类

目录 内容说明 章节内容 一、 File类 内容说明 结合章节内容重点难点,会对重要知识点进行扩展,以及做示例说明等,以便更好理解重点难点 章节内容 一、 File类 1、文件与目录的描述类——File ✔️ File类并不用来进行文件的读/写操作,并未涉及到写入或读取文件内容的…...

测试面试题集锦(四)| Linux 与 Python 编程篇(附答案)

本系列文章总结归纳了一些软件测试工程师常见的面试题&#xff0c;主要来源于个人面试遇到的、网络搜集&#xff08;完善&#xff09;、工作日常讨论等&#xff0c;分为以下十个部分&#xff0c;供大家参考。如有错误的地方&#xff0c;欢迎指正。有更多的面试题或面试中遇到的…...

pytorch中的矩阵乘法

1. 运算符介绍 关于运算&#xff0c;*运算&#xff0c;torch.mul(), torch.mm(), torch.mv(), tensor.t() 和 *代表矩阵的两种相乘方式&#xff1a; 表示常规的数学上定义的矩阵相乘&#xff1b; *表示两个矩阵对应位置处的两个元素相乘。 1.1 矩阵点乘 *和torch.mul()等同…...

Java--Stream流详解

Stream是Java 8 API添加的一个新的抽象&#xff0c;称为流Stream&#xff0c;以一种声明性方式处理数据集合&#xff08;侧重对于源数据计算能力的封装&#xff0c;并且支持序列与并行两种操作方式&#xff09; Stream流是从支持数据处理操作的源生成的元素序列&#xff0c;源可…...

[PHP]ShopXO企业级B2C免费开源商城系统 v2.3.1

ShopXO 企业级B2C免费开源电商系统&#xff01; 求实进取、创新专注、自主研发、国内领先企业级B2C电商系统解决方案。 遵循Apache2开源协议发布&#xff0c;无需授权、可商用、可二次开发、满足99%的电商运营需求。 PCH5、支付宝小程序、微信小程序、百度小程序、头条&抖音…...

Python基础入门系列详解20篇

Python基础入门&#xff08;1&#xff09;----Python简介 Python基础入门&#xff08;2&#xff09;----安装Python环境&#xff08;Windows、MacOS、CentOS、Ubuntu&#xff09; Python基础入门&#xff08;3&#xff09;----Python基础语法&#xff1a;解释器、标识符、关键…...

P02项目(学习)

★ P02项目 项目描述&#xff1a;安全操作项目旨在提高医疗设备的安全性&#xff0c;特别是在医生离开操作屏幕时&#xff0c;以减少非授权人员的误操作风险。为实现这一目标&#xff0c;我们采用多层次的保护措施&#xff0c;包括人脸识别、姿势检测以及二维码识别等技术。这些…...

pandas 笔记:get_dummies分类变量one-hot化

1 函数介绍 pandas.get_dummies 是 pandas 库中的一个函数&#xff0c;它用于将分类变量转换为哑变量/指示变量。所谓的哑变量&#xff0c;就是将分类变量的每一个不同的值转换为一个新的0/1变量。在输出的DataFrame中&#xff0c;每一列都以该值的名称命名 pandas.get_dummi…...

PTE作文练习(一)

目录 65分备考建议 WE模版 范文 Supporting ideas: SWT 65分备考建议 RA重在多听标准的正确的示范&#xff0c;RS重在抓大放小&#xff0c;WFD重在整理错题&#xff0c;以及反反复复的车轮战&#xff0c;FIBRW重在“以对代记” 就是直接看答案&#xff0c;节约时间&#…...

如何做到一套FPGA工程无缝兼容两款不同的板卡?

试想这样一种场景,有两款不同的FPGA板卡,它们的功能代码90%都是一样的,但是两个板卡的管脚分配完全不同,一般情况下,我们需要设计两个工程,两套代码,之后还需要一直维护两个版本。 那么有没有一种自动化的方式,实现一个工程,编译出一个程序文件,下载到这两个不同的板…...

VSCode修改主题为Eclipse 绿色护眼模式

前言 从参加开发以来&#xff0c;一直使用eclipse进行开发&#xff0c;基本官方出新版本&#xff0c;我都会更新。后来出来很多其他的IDE工具&#xff0c;我也尝试了&#xff0c;但他们的主题都把我劝退了&#xff0c;黑色主题是谁想出来&#xff1f;&#x1f602; 字体小的时…...

conan和cmake编译器版本不匹配问题解决

conan和cmake编译器版本不匹配问题解决 1 问题现象2 解决方法2.1 在CMakeLists.txt禁止编译器检查2.1.1 修改方式 2.2 探查问题出现的根本原因2.2.1 安装升级gcc2.2.2 安装升级g 注 执行环境&#xff1a;ubuntu 1 问题现象 conan要求的编译器版本和cmake检测到的当前的编译器…...

float单精度浮点数如何在计算机中存储

文章目录 1 float型数据组成2 实际举例3 代码测试4 写在最后 1 float型数据组成 按照IEEE浮点标准存储浮点数时&#xff0c;一个float型的值由1个符号位&#xff08;最左边的位或最高有效位&#xff09;、8个指数位以及23个小数位依次组成: 符号位为0时表示正数&#xff0c;为1…...

机器视觉在虚拟现实与增强现实中的作用

机器视觉在虚拟现实&#xff08;VR&#xff09;和增强现实&#xff08;AR&#xff09;中发挥着至关重要的作用。这些技术的核心是计算机视觉领域&#xff0c;重点是让计算机具有“看到”和理解周围世界的能力。 在虚拟现实中&#xff0c;计算机视觉用于创建和处理用户所见的虚…...

红黑数原理及存在原因

我红黑树那么牛&#xff0c;你们为什么不用&#xff1f;_哔哩哔哩_bilibili 面试时经常会被问到红黑树&#xff0c;它到底有什么优点呢&#xff1f; 对于查找数据&#xff0c;数组二分查询速度最快&#xff0c;时间复杂度为O(logN)。但是如果增加和删除数据&#xff0c;数组就…...

Ansible入门—安装部署及各个模块应用案例(超详细)

目录 前言 一、环境概况 修改主机名&#xff08;可选项&#xff09; 二、安装部署 1.安装epel扩展源 2.安装Ansible 3.修改Ansible的hosts文件 4.生成密钥 三、Ansible模块使用介绍 Command模块 Shell模块 User模块 Copy模块 File模块 Hostname模块 Yum模块 Se…...

Spring Boot 3系列之-启动类详解

Spring Boot是一个功能强大、灵活且易于使用的框架&#xff0c;它极大地简化了Spring应用程序的开发和部署流程&#xff0c;使得开发人员能够更专注于业务逻辑的实现。在我们的Spring Boot 3系列之一&#xff08;初始化项目&#xff09;文章中&#xff0c;我们使用了Spring官方…...

智能AI电话机器人系统的识别能力现状与发展水平

一、引言 随着人工智能技术的飞速发展&#xff0c;AI电话机器人系统已经从简单的自动应答工具演变为具备复杂交互能力的智能助手。这类系统结合了语音识别、自然语言处理、情感计算和机器学习等多项前沿技术&#xff0c;在客户服务、营销推广、信息查询等领域发挥着越来越重要…...

[ACTF2020 新生赛]Include 1(php://filter伪协议)

题目 做法 启动靶机&#xff0c;点进去 点进去 查看URL&#xff0c;有 ?fileflag.php说明存在文件包含&#xff0c;原理是php://filter 协议 当它与包含函数结合时&#xff0c;php://filter流会被当作php文件执行。 用php://filter加编码&#xff0c;能让PHP把文件内容…...

es6+和css3新增的特性有哪些

一&#xff1a;ECMAScript 新特性&#xff08;ES6&#xff09; ES6 (2015) - 革命性更新 1&#xff0c;记住的方法&#xff0c;从一个方法里面用到了哪些技术 1&#xff0c;let /const块级作用域声明2&#xff0c;**默认参数**&#xff1a;函数参数可以设置默认值。3&#x…...

数据库——redis

一、Redis 介绍 1. 概述 Redis&#xff08;Remote Dictionary Server&#xff09;是一个开源的、高性能的内存键值数据库系统&#xff0c;具有以下核心特点&#xff1a; 内存存储架构&#xff1a;数据主要存储在内存中&#xff0c;提供微秒级的读写响应 多数据结构支持&…...

HTTPS证书一年多少钱?

HTTPS证书作为保障网站数据传输安全的重要工具&#xff0c;成为众多网站运营者的必备选择。然而&#xff0c;面对市场上种类繁多的HTTPS证书&#xff0c;其一年费用究竟是多少&#xff0c;又受哪些因素影响呢&#xff1f; 首先&#xff0c;HTTPS证书通常在PinTrust这样的专业平…...

鸿蒙Navigation路由导航-基本使用介绍

1. Navigation介绍 Navigation组件是路由导航的根视图容器&#xff0c;一般作为Page页面的根容器使用&#xff0c;其内部默认包含了标题栏、内容区和工具栏&#xff0c;其中内容区默认首页显示导航内容&#xff08;Navigation的子组件&#xff09;或非首页显示&#xff08;Nav…...

【笔记】AI Agent 项目 SUNA 部署 之 Docker 构建记录

#工作记录 构建过程记录 Microsoft Windows [Version 10.0.27871.1000] (c) Microsoft Corporation. All rights reserved.(suna-py3.12) F:\PythonProjects\suna>python setup.py --admin███████╗██╗ ██╗███╗ ██╗ █████╗ ██╔════╝…...

Python爬虫(52)Scrapy-Redis分布式爬虫架构实战:IP代理池深度集成与跨地域数据采集

目录 一、引言&#xff1a;当爬虫遭遇"地域封锁"二、背景解析&#xff1a;分布式爬虫的两大技术挑战1. 传统Scrapy架构的局限性2. 地域限制的三种典型表现 三、架构设计&#xff1a;Scrapy-Redis 代理池的协同机制1. 分布式架构拓扑图2. 核心组件协同流程 四、技术实…...

统计按位或能得到最大值的子集数目

我们先来看题目描述&#xff1a; 给你一个整数数组 nums &#xff0c;请你找出 nums 子集 按位或 可能得到的 最大值 &#xff0c;并返回按位或能得到最大值的 不同非空子集的数目 。 如果数组 a 可以由数组 b 删除一些元素&#xff08;或不删除&#xff09;得到&#xff0c;…...

实现p2p的webrtc-srs版本

1. 基本知识 1.1 webrtc 一、WebRTC的本质&#xff1a;实时通信的“网络协议栈”类比 将WebRTC类比为Linux网络协议栈极具洞察力&#xff0c;二者在架构设计和功能定位上高度相似&#xff1a; 分层协议栈架构 Linux网络协议栈&#xff1a;从底层物理层到应用层&#xff08;如…...