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

五分钟,Docker安装flink,并使用flinksql消费kafka数据

1、拉取flink镜像,创建网络

docker pull flink
docker network create flink-network

2、创建 jobmanager

# 创建 JobManager docker run \-itd \--name=jobmanager \--publish 8081:8081 \--network flink-network \--env FLINK_PROPERTIES="jobmanager.rpc.address: jobmanager" \flink:latest jobmanager 

3、创建 taskmanager

# 创建 TaskManager docker run \-itd \--name=taskmanager \--network flink-network \--env FLINK_PROPERTIES="jobmanager.rpc.address: jobmanager" \flink:latest taskmanager 

4、访问 http://localhost:8081/

在这里插入图片描述

4.1 修改Task Slots

默认的Slots num是1,我们可以修改为5:
修改的目录是jobmanager和taskmanager的/opt/flink/confflink-conf.yaml文件:
在这里插入图片描述
在这里插入图片描述
修改taskmanager.numberOfTaskSlots:即可。
注意:默认的docker容器中没有vi/vim命令,可以使用docker cp命令,复制出来修改,然后在复制回去,如下:

docker cp taskmanager:/opt/flink/conf/flink-conf.yaml .
docker cp flink-conf.yaml taskmanager:/opt/flink/conf/

在这里插入图片描述

5、通过flinksql消费Kafka

确保有一个可用的kafka,如果没有,可以五分钟内,Docker搭建一个
Docker安装kafka 3.5
并且通过python,简单写一个生产者
Python生产、消费Kafka

5.1 导入flink-sql-connector-kafka jar包

顾名思义,用于连接flinksql和kafka。
进入flink

docker exec -it jobmanager /bin/bash

进入 flink的bin目录

cd /opt/flink/bin

查看flink版本:

flink --version

可以看出,我的版本是1.18.0
在这里插入图片描述
根据自己的flink版本,下载对应的 flink-sql-connector-kafka jar包
https://mvnrepository.com/artifact/org.apache.flink/flink-sql-connector-kafka
因为我是1.18.0,所以选择下图的版本包:
在这里插入图片描述
点进去进行下载:
在这里插入图片描述

将下载的jar包,分别在jobmanager,taskmanager /opt/flink/lib目录下,注意,是两个都要放,如下图:
在这里插入图片描述
可以使用docker cp test.txt jobmanager:/opt/flink/lib命令,用户宿主机和docker容器文件传输。把test.txt换成对应的jar包即可

5.2 flinksql消费kafka

进入jobmanager中,执行

cd /opt/flink/bin
sql-client.sh

在这里插入图片描述
在这里插入图片描述
Flink SQL执行以下语句:

CREATE TABLE KafkaTable (`count_num` STRING,`ts` TIMESTAMP(3) METADATA FROM 'timestamp'
) WITH ('connector' = 'kafka','topic' = 'kafka_demo','properties.bootstrap.servers' = '192.168.10.15:9092','properties.group.id' = 'testGroup','scan.startup.mode' = 'earliest-offset','format' = 'json'
);show tables;
select * from KafkaTable;

可以看到Flink在消费kafka数据,如下图:
在这里插入图片描述

相关文章:

五分钟,Docker安装flink,并使用flinksql消费kafka数据

1、拉取flink镜像,创建网络 docker pull flink docker network create flink-network2、创建 jobmanager # 创建 JobManager docker run \-itd \--namejobmanager \--publish 8081:8081 \--network flink-network \--env FLINK_PROPERTIES"jobmanager.rpc.ad…...

【小聆送书第一期】让架构师的成神之路温暖你这个不景气的冬天

🌈个人主页:聆风吟 🔥系列专栏:网络奇遇记、数据结构 🔖少年有梦不应止于心动,更要付诸行动。 文章目录 📋前言 书籍一览 ⛳️书籍一⛳️书籍二⛳️书籍三⛳️书籍四⛳️书籍五⛳️书籍六⛳️书…...

网页爬虫反扒措施有哪些?

爬虫之常见的反扒 cookies 一般用requests直接请求网址的时候有时候可能会遇到反扒措施,这时候可以考虑一下加上user-agent伪装成浏览器;也可能有登录限制,这时候cookies就有用处了 浏览器中的cookie是保存我们的账号数据和访问记录&#…...

C#实现批量生成二维码

相信大家都使用过草料二维码生成器&#xff0c;单独生成二维码可以&#xff0c;但是批量生成二维码就需要收费了。既然要收费&#xff0c;那就自己写一个。 接口采用导入Excel文件生成二维码&#xff0c;首先需要读取Excel的数据&#xff0c;方法如下所示&#xff1a; /// <…...

3种在ArcGIS Pro中制作山体阴影的方法

山体阴影可以更直观的展现地貌特点&#xff0c;表达真实的地形&#xff0c;这里为大家介绍一下在ArcGIS Pro中制作山体阴影的方法&#xff0c;希望能对你有所帮助。 数据来源 本教程所使用的数据是从水经微图中下载的DEM数据&#xff0c;除了DEM数据&#xff0c;常见的GIS数据…...

【ChatGLM2-6B】Docker下部署及微调

【ChatGLM2-6B】小白入门及Docker下部署 一、简介1、ChatGLM2是什么2、组成部分3、相关地址 二、基于Docker安装部署1、前提2、CentOS7安装NVIDIA显卡驱动1&#xff09;查看服务器版本及显卡信息2&#xff09;相关依赖安装3&#xff09;显卡驱动安装 2、 CentOS7安装NVIDIA-Doc…...

输入两个整数,输出它们的乘积。 ← Python 及 C++ 代码比较

【题目描述】 输入两个整数&#xff0c;输出它们的乘积。【Python代码】 x,ymap(int,input().split()) print(x*y) 【C代码】 #include<bits/stdc.h> using namespace std;int x,y; int main() {cin>>x>>y;cout<<x*y<<endl;return 0; }/* in:…...

C语言——从键盘输人一个表示年份的整数,判断该年份是否为闰年,并显示判断结果。

#define _CRT_SECURE_NO_WARNINGS 1#include<stdio.h> int main() {int year 0;printf("请输入年份&#xff1a;");scanf("%d",&year);if((year%4 0) && (year%100!0) || (year%400 0)){printf("%d是闰年\n",year);}else{p…...

出于隐私和安全的考虑,有时需要从谷歌删除你的个人数据,有两种方法

如果你是公众人物、企业或拥有个人品牌的人&#xff0c;那么拥有在线形象很重要。然而&#xff0c;你可能会发现&#xff0c;通过谷歌搜索&#xff0c;陌生人可以获得你的个人信息&#xff0c;如联系方式、地址和财务信息&#xff0c;这会让你感到不安。 幸运的是&#xff0c;…...

【同一局域网下】两台电脑之间互ping

两台电脑互ping 首先需要连接同一网咯关闭需要ping的电脑的防火墙 关闭防火墙步骤&#xff08;以win11系统为例&#xff09;&#xff1a; 设置 --> 隐私和安全性 --> Windows 安全中心 打开Windows安全中心 防火墙和网络保护 --> 选择正在使用的网络 关闭 ping其他…...

【精选】Ajax技术知识点合集

Ajax技术详解 Ajax简介 Ajax 即“Asynchronous Javascript And XML”&#xff08;异步 JavaScript 和 XML&#xff09;&#xff0c;是指一种创建 交互式、快速动态应用的网页开发技术&#xff0c;无需重新加载整个网页的情况下&#xff0c;能够更新页面局 部数据的技术。通过在…...

智能优化算法应用:基于水循环算法无线传感器网络(WSN)覆盖优化 - 附代码

智能优化算法应用&#xff1a;基于水循环算法无线传感器网络(WSN)覆盖优化 - 附代码 文章目录 智能优化算法应用&#xff1a;基于水循环算法无线传感器网络(WSN)覆盖优化 - 附代码1.无线传感网络节点模型2.覆盖数学模型及分析3.水循环算法4.实验参数设定5.算法结果6.参考文献7.…...

java-netty知识点笔记和注意事项

如何获取ctx的id 使用ctx.ctx.toString()就可以了 public void channelRead(ChannelHandlerContext ctx, Object msg) {//传来的消息包装成字节缓冲区String byteBuf (String) msg; // ByteBuf byteBuf (ByteBuf) msg;//Netty提供了字节缓冲区的toString方法&#xff…...

英伟达不同系列GPU介绍

英伟达有以下几个系列的产品线&#xff0c;并介绍它们的特点和主要应用领域&#xff1a; 1. GeForce系列&#xff08;G系列&#xff09;&#xff1a; - 特点&#xff1a;GeForce系列是英伟达主打的消费级GPU产品线&#xff0c;注重提供高性能的图形处理能力和游戏特性。它们…...

C语言——I /深入理解指针(二)

一、数组名的理解 int arr[10] {1,2,3,4,5,6,7,8,9,10}; int *p &arr[0];这⾥我们使⽤ &arr[0] 的⽅式拿到了数组第⼀个元素的地址&#xff0c;但是其实数组名本来就是地址&#xff0c;⽽且 是数组⾸元素的地址&#xff0c;我们来做个测试。 #include <stdio.…...

MySQL使用函数和存储过程实现:向数据表快速插入大量测试数据

实现过程 1.创建表 CREATE TABLE user_info (id INT(11) NOT NULL AUTO_INCREMENT,name VARCHAR(20) DEFAULT NULL,age INT(3) DEFAULT NULL,pwd VARCHAR(20) DEFAULT NULL,phone_number VARCHAR(11) DEFAULT NULL,email VARCHAR(255) DEFAULT NULL,address VARCHAR(255) DEF…...

力扣labuladong——一刷day59

提示&#xff1a;文章写完后&#xff0c;目录可以自动生成&#xff0c;如何生成可参考右边的帮助文档 文章目录 前言一、力扣549. 二叉树中最长的连续序列二、力扣1325. 删除给定值的叶子节点 前言 像求和、求高度这种基本的二叉树函数很容易写&#xff0c;有时候只要在它们的后…...

接口性能测试 —— Jmeter并发与持续性压测

接口压测的方式&#xff1a; 1、同时并发&#xff1a;设置线程组、执行时间、循环次数&#xff0c;这种方式可以控制接口请求的次数 2、持续压测&#xff1a;设置线程组、循环次数&#xff0c;勾选“永远”&#xff0c;调度器&#xff08;持续时间&#xff09;&#xff0c;这种…...

redis报错3

INFO: Initializing SpringDispatcherServletdispatcherServlet...

Proteus的网络标号与总线

Proteus为了减少过多、复杂的连线&#xff0c;可以使用网络标号与总线配合使用。 Proteus的导线上添加了网络标号&#xff0c;意味着在Proteus上相同的网络标号是连在一起的&#xff0c;所说在图纸上看不出来。 如下图是比较好的Proteus中使用总线的绘制的图纸。可以效仿着画…...

Docker 离线安装指南

参考文章 1、确认操作系统类型及内核版本 Docker依赖于Linux内核的一些特性&#xff0c;不同版本的Docker对内核版本有不同要求。例如&#xff0c;Docker 17.06及之后的版本通常需要Linux内核3.10及以上版本&#xff0c;Docker17.09及更高版本对应Linux内核4.9.x及更高版本。…...

UE5 学习系列(三)创建和移动物体

这篇博客是该系列的第三篇&#xff0c;是在之前两篇博客的基础上展开&#xff0c;主要介绍如何在操作界面中创建和拖动物体&#xff0c;这篇博客跟随的视频链接如下&#xff1a; B 站视频&#xff1a;s03-创建和移动物体 如果你不打算开之前的博客并且对UE5 比较熟的话按照以…...

Go 语言接口详解

Go 语言接口详解 核心概念 接口定义 在 Go 语言中&#xff0c;接口是一种抽象类型&#xff0c;它定义了一组方法的集合&#xff1a; // 定义接口 type Shape interface {Area() float64Perimeter() float64 } 接口实现 Go 接口的实现是隐式的&#xff1a; // 矩形结构体…...

Python爬虫(一):爬虫伪装

一、网站防爬机制概述 在当今互联网环境中&#xff0c;具有一定规模或盈利性质的网站几乎都实施了各种防爬措施。这些措施主要分为两大类&#xff1a; 身份验证机制&#xff1a;直接将未经授权的爬虫阻挡在外反爬技术体系&#xff1a;通过各种技术手段增加爬虫获取数据的难度…...

如何理解 IP 数据报中的 TTL?

目录 前言理解 前言 面试灵魂一问&#xff1a;说说对 IP 数据报中 TTL 的理解&#xff1f;我们都知道&#xff0c;IP 数据报由首部和数据两部分组成&#xff0c;首部又分为两部分&#xff1a;固定部分和可变部分&#xff0c;共占 20 字节&#xff0c;而即将讨论的 TTL 就位于首…...

libfmt: 现代C++的格式化工具库介绍与酷炫功能

libfmt: 现代C的格式化工具库介绍与酷炫功能 libfmt 是一个开源的C格式化库&#xff0c;提供了高效、安全的文本格式化功能&#xff0c;是C20中引入的std::format的基础实现。它比传统的printf和iostream更安全、更灵活、性能更好。 基本介绍 主要特点 类型安全&#xff1a…...

springboot 日志类切面,接口成功记录日志,失败不记录

springboot 日志类切面&#xff0c;接口成功记录日志&#xff0c;失败不记录 自定义一个注解方法 import java.lang.annotation.ElementType; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target;/***…...

GraphQL 实战篇:Apollo Client 配置与缓存

GraphQL 实战篇&#xff1a;Apollo Client 配置与缓存 上一篇&#xff1a;GraphQL 入门篇&#xff1a;基础查询语法 依旧和上一篇的笔记一样&#xff0c;主实操&#xff0c;没啥过多的细节讲解&#xff0c;代码具体在&#xff1a; https://github.com/GoldenaArcher/graphql…...

Linux安全加固:从攻防视角构建系统免疫

Linux安全加固:从攻防视角构建系统免疫 构建坚不可摧的数字堡垒 引言:攻防对抗的新纪元 在日益复杂的网络威胁环境中,Linux系统安全已从被动防御转向主动免疫。2023年全球网络安全报告显示,高级持续性威胁(APT)攻击同比增长65%,平均入侵停留时间缩短至48小时。本章将从…...

【iOS】 Block再学习

iOS Block再学习 文章目录 iOS Block再学习前言Block的三种类型__ NSGlobalBlock____ NSMallocBlock____ NSStackBlock__小结 Block底层分析Block的结构捕获自由变量捕获全局(静态)变量捕获静态变量__block修饰符forwarding指针 Block的copy时机block作为函数返回值将block赋给…...