flink StreamGraph 构造flink任务
文章目录
- 背景
- 主要步骤
- 代码
背景
通常使用flink 提供的高级算子来编写flink 任务,对底层不是很了解,尤其是如何生成作业图的细节
下面通过构造一个有向无环图,来实际看一下
主要步骤
1.增加source
2.增加operator
3. 增加一条边,连接source和operator
4. 增加sink
5. 增加一条边,连接operator和sink
代码
// Step 1: Create basic configurationsConfiguration configuration = new Configuration();ExecutionConfig executionConfig = new ExecutionConfig();CheckpointConfig checkpointConfig = new CheckpointConfig();SavepointRestoreSettings savepointRestoreSettings = SavepointRestoreSettings.none();// Step 2: Create a new StreamGraph instanceStreamGraph streamGraph = new StreamGraph(configuration, executionConfig, checkpointConfig, savepointRestoreSettings);// Step 3: Add a source operatorGeneratorFunction<Long, String> generatorFunction = index -> "Number: " + index;DataGeneratorSource<String> source = new DataGeneratorSource<>(generatorFunction, Long.MAX_VALUE, RateLimiterStrategy.perSecond(1), Types.STRING);SourceOperatorFactory<String> sourceOperatorFactory = new SourceOperatorFactory<>(source, WatermarkStrategy.noWatermarks());streamGraph.addSource(1, "sourceNode", "sourceDescription", sourceOperatorFactory, TypeInformation.of(String.class), TypeInformation.of(String.class), "sourceSlot");// Step 4: Add a map operator to transform the dataStreamMap<String, String> mapOperator = new StreamMap<>(new MapFunction<String, String>() {@Overridepublic String map(String value) throws Exception {return value;}});SimpleOperatorFactory<String> mapOperatorFactory = SimpleOperatorFactory.of(mapOperator);streamGraph.addOperator(2, "mapNode", "mapDescription", mapOperatorFactory, TypeInformation.of(String.class), TypeInformation.of(String.class), "mapSlot");// Step 5: Connect source and map operatorstreamGraph.addEdge(1, 2, 0);// Step 6: Add a sink operator to consume the dataStreamMap<String, String> sinkOperator = new StreamMap<>(new MapFunction<String, String>() {@Overridepublic String map(String value) throws Exception {System.out.println(value);return value;}});SimpleOperatorFactory<String> sinkOperatorFactory = SimpleOperatorFactory.of(sinkOperator);streamGraph.addSink(3, "sinkNode", "sinkDescription", sinkOperatorFactory, TypeInformation.of(String.class), TypeInformation.of(String.class), "sinkSlot");// Step 7: Connect map and sink operatorstreamGraph.addEdge(2, 3, 0);streamGraph.setTimeCharacteristic(TimeCharacteristic.ProcessingTime);streamGraph.setMaxParallelism(1,1);streamGraph.setMaxParallelism(2,1);streamGraph.setMaxParallelism(3,1);streamGraph.setGlobalStreamExchangeMode(GlobalStreamExchangeMode.ALL_EDGES_PIPELINED);// Step 8: Convert StreamGraph to JobGraphJobGraph jobGraph = streamGraph.getJobGraph();// Step 9: Set up a MiniCluster for local executionMiniClusterConfiguration miniClusterConfig = new MiniClusterConfiguration.Builder().setNumTaskManagers(10).setNumSlotsPerTaskManager(10).build();MiniCluster miniCluster = new MiniCluster(miniClusterConfig);// Step 10: Start the MiniClusterminiCluster.start();// Step 11: Submit the job to the MiniClusterJobExecutionResult result = miniCluster.executeJobBlocking(jobGraph);System.out.println("Job completed with result: " + result);// Step 12: Stop the MiniClusterminiCluster.close();
相关文章:

flink StreamGraph 构造flink任务
文章目录 背景主要步骤代码 背景 通常使用flink 提供的高级算子来编写flink 任务,对底层不是很了解,尤其是如何生成作业图的细节 下面通过构造一个有向无环图,来实际看一下 主要步骤 1.增加source 2.增加operator 3. 增加一条边࿰…...

【51单片机】LCD1602液晶显示屏
学习使用的开发板:STC89C52RC/LE52RC 编程软件:Keil5 烧录软件:stc-isp 开发板实图: 文章目录 LCD1602存储结构时序结构 编码 —— 显示字符、数字 LCD1602 LCD1602(Liquid Crystal Display)液晶显示屏是…...

理解 HTML5 Canvas 中逻辑像素与物理像素的关系
理解 HTML5 Canvas 中逻辑像素与物理像素的关系 在使用 HTML5 Canvas 时,开发者经常会遇到一个困惑:为什么鼠标的 offsetX 和 offsetY 和我绘制的图形坐标对不上?这通常是因为 Canvas 的逻辑像素大小和物理像素大小不一致。本文将详细解释这…...

7.揭秘C语言输入输出内幕:printf与scanf的深度剖析
揭秘C语言输入输出内幕:printf与scanf的深度剖析 C语言往期系列文章目录 往期回顾: VS 2022 社区版C语言的安装教程,不要再卡在下载0B/s啦C语言入门:解锁基础概念,动手实现首个C程序C语言概念之旅:解锁关…...

数据分析-系统认识数据分析
目录 数据分析的全貌 观测 实验 应用 数据分析的全貌 观测 实验 应用...

蓝桥杯介绍
赛事背景与历程 自2009年举办以来,蓝桥杯已经连续举行了多届,成为国内领先的信息技术赛事。2022年,蓝桥杯被教育部确定为2022—2025学年面向中小学生的全国性竞赛活动,并入选国家级A类学科竞赛。 参赛对象与组别 蓝桥杯的参赛对…...

鸿蒙加载网络图片并转换成PixelMap
鸿蒙加载网络图片并转换成PixelMap 参考文档 基于API12. 有一些图片功能需要使用 PixelMap 类型的参数,但是使用Image组件之类的时候无法获取到 PixelMap 类型数据。 因此只能是把图片下载下来然后加在并转换一下。 实现方式 一下封装了一个函数。使用的 rcp 模…...

hive搭建
1.准备环境 三台节点主机已安装hadoopmysql数据库 2.环境 2.1修改三台节点上hadoop的core-site.xml <!-- 配置 HDFS 允许代理任何主机和组 --> <property><name>hadoop.proxyuser.hadoop.hosts</name><value>*</value> </property&…...

51c扩散模型~合集1
我自己的原文哦~ https://blog.51cto.com/whaosoft/11541675 #Diffusion Forcing 无限生成视频,还能规划决策,扩散强制整合下一token预测与全序列扩散 当前,采用下一 token 预测范式的自回归大型语言模型已经风靡全球,同时互联…...

从零开始深度学习:全连接层、损失函数与梯度下降的详尽指南
引言 在深度学习的领域,全连接层、损失函数与梯度下降是三块重要的基石。如果你正在踏上深度学习的旅程,理解它们是迈向成功的第一步。这篇文章将从概念到代码、从基础到进阶,详细剖析这三个主题,帮助你从小白成长为能够解决实际…...

Liebherr利勃海尔 EDI 需求分析
Liebherr 使用 EDI 技术来提高业务流程的效率、降低错误率、加快数据交换速度,并优化与供应商、客户和其他合作伙伴之间的业务沟通。通过 EDI,Liebherr 实现了与全球交易伙伴的自动化数据交换,提升了供应链管理和订单处理的透明度。 Liebher…...

java小练习
小练1.用while语句计算11/2!1/3!1/4!...1/20!的和 public class test_11_17_2 {public static void main(String[] args) {double sum 0;double item 1;int n 20;int i 1;while(i<n){sum item;i i1;item item*(1.0/i);}System.out.println(sum);} } 小练2.计算88888…...

go语言中的占位符有哪些
在Go语言中,占位符主要用于格式化字符串输出,特别是在使用fmt包中的Printf系列函数时。以下是Go语言中常用的占位符: %v:代表值的默认格式,对于字符串是直接输出,对于整型是十进制形式。%v:扩展…...

基于Windows安装opus python库
项目中需要用到一些opus格式的编解码功能,找到网上有opus的开源库。网址:Opus Codec 想着人生苦短,没想到遇上了错误!在这里记录一下过程 过程 安装python库 pip3 install opuslib验证 >>> import opuslib Tracebac…...

【设计模式】行为型模式(五):解释器模式、访问者模式、依赖注入
《设计模式之行为型模式》系列,共包含以下文章: 行为型模式(一):模板方法模式、观察者模式行为型模式(二):策略模式、命令模式行为型模式(三):责…...

使用nossl模式连接MySQL数据库详解
使用nossl模式连接MySQL数据库详解 摘要一、引言二、nossl模式概述2.1 SSL与nossl模式的区别2.2 选择nossl模式的场景三、在nossl模式下连接MySQL数据库3.1 准备工作3.2 C++代码示例3.3 代码详解3.3.1 初始化MySQL连接对象3.3.2 连接到MySQL数据库3.3.3 执行查询操作3.3.4 处理…...

【MySQL】ubantu 系统 MySQL的安装与免密码登录的配置
🍑个人主页:Jupiter. 🚀 所属专栏:MySQL初阶探索:构建数据库基础 欢迎大家点赞收藏评论😊 目录 📚mysql的安装📕MySQL的登录🌏MySQL配置免密码登录 📚mysql的…...

高级 SQL 技巧讲解
大家好,我是程序员小羊! 前言: SQL(结构化查询语言)是管理和操作数据库的核心工具。从基本的查询语句到复杂的数据处理,掌握高级 SQL 技巧不仅能显著提高数据分析的效率,还能解决业务中的复…...

浅论AI大模型在电商行业的发展未来
随着人工智能(AI)技术的快速发展,AI大模型在电商行业中扮演着越来越重要的角色。本文旨在探讨AI大模型如何赋能电商行业,包括提升销售效率、优化用户体验、增强供应链管理等方面。通过分析AI大模型在电商领域的应用案例和技术进展…...

【python笔记03】《类》
文章目录 面向对象基本概念对象的概念类的概念 类的定义类的创建(实例的模板)类的实例化--获取对象对象方法中的self关键字面试题请描述什么是对象,什么是类。请观阅读如下代码,判断是否能正常运行,如果不能正常运行&a…...

Flutter 应用在真机上调试的流程
在真机上调试 Flutter 应用的方法有很多,可以使用 USB 数据线连接设备到电脑进行调试,也可以通过无线方式进行 Flutter 真机调试。 1. 有线调试 设备准备 启用开发者模式: Android:进入 设置 > 关于手机,连续点击…...

以太坊基础知识结构详解
以太坊的历史和发展 初创阶段 2013年:Vitalik Buterin 发表了以太坊白皮书,提出了一个通用的区块链平台,不仅支持比特币的货币功能,还能支持更复杂的智能合约。2014年:以太坊项目启动,进行了首次ICO&…...

安全见闻(完整版)
目录 安全见闻1 编程语言和程序 编程语言 函数式编程语言: 数据科学和机器学习领域: Web 全栈开发: 移动开发: 嵌入式系统开发: 其他: 编程语言的方向: 软件程序 操作系统 硬件设备…...

LeetCode100之反转链表(206)--Java
1.问题描述 给你单链表的头节点 head ,请你反转链表,并返回反转后的链表 示例1 输入:head [1,2,3,4,5] 输出:[5,4,3,2,1] 示例2 输入:head [1,2] 输出:[2,1] 示例3 输入:head [] 输…...

牛客周赛第一题2024/11/17日
链接:登录—专业IT笔试面试备考平台_牛客网 来源:牛客网 时间限制:C/C/Rust/Pascal 1秒,其他语言2秒 空间限制:C/C/Rust/Pascal 256 M,其他语言512 M 64bit IO Format: %lld 题目描述 小红这天来到了三…...

麒麟Server下安装东方通TongLINK/Q
环境 系统:麒麟Server SP3 2403 应用:TLQ8.1(Install_TLQ_Standard_Linux2.6.32_x86_64_8.1.17.0.tar.gz) 安装Server 将文件解压到/usr/local/tlq。 cd /opt/tlq/ mkdir /usr/local/tlq/ tar -zxvf Install_TLQ_Standard_Linux2.6.32_x86_64_8.1.1…...

BERT的中文问答系统33
我们在现有的代码基础上增加网络搜索的功能。我们使用 requests 和 BeautifulSoup 来从百度搜索结果中提取信息。以下是完整的代码,包括项目结构、README.md 文件以及所有必要的代码。 项目结构 xihe241117/ ├── data/ │ └── train_data.jsonl ├── lo…...

Ubuntu下的Eigen库的安装及基本使用教程
一、Eigen库介绍 简介 Eigen [1]目前最新的版本是3.4,除了C标准库以外,不需要任何其他的依赖包。Eigen使用的CMake建立配置文件和单元测试,并自动安装。如果使用Eigen库,只需包特定模块的的头文件即可。 基本功能 Eigen适用范…...

【spring 】Spring Cloud Gateway 的Filter学习
介绍和使用场景 Spring Cloud Gateway 是一个基于 Spring Framework 5 和 Project Reactor 的 API 网关,它旨在为微服务架构提供一种简单而有效的方式来处理请求路由、过滤、限流等功能。在 Spring Cloud Gateway 中,Filter 扮演着非常重要的角色&#…...

每秒交易数(Transactions Per Second:TPS)详细拆解
每秒交易数(TPS)是指计算机网络每秒可以处理的交易数量。TPS是衡量不同区块链和其他计算机系统速度的关键指标。然而,TPS并不是用来衡量区块链速度的唯一指标。许多人认为,虽然TPS很重要,但最终性实际上是一个更重要的…...