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

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 任务&#xff0c;对底层不是很了解&#xff0c;尤其是如何生成作业图的细节 下面通过构造一个有向无环图&#xff0c;来实际看一下 主要步骤 1.增加source 2.增加operator 3. 增加一条边&#xff0…...

【51单片机】LCD1602液晶显示屏

学习使用的开发板&#xff1a;STC89C52RC/LE52RC 编程软件&#xff1a;Keil5 烧录软件&#xff1a;stc-isp 开发板实图&#xff1a; 文章目录 LCD1602存储结构时序结构 编码 —— 显示字符、数字 LCD1602 LCD1602&#xff08;Liquid Crystal Display&#xff09;液晶显示屏是…...

理解 HTML5 Canvas 中逻辑像素与物理像素的关系

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

7.揭秘C语言输入输出内幕:printf与scanf的深度剖析

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

数据分析-系统认识数据分析

目录 数据分析的全貌 观测 实验 应用 数据分析的全貌 观测 实验 应用...

蓝桥杯介绍

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

鸿蒙加载网络图片并转换成PixelMap

鸿蒙加载网络图片并转换成PixelMap 参考文档 基于API12. 有一些图片功能需要使用 PixelMap 类型的参数&#xff0c;但是使用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 无限生成视频&#xff0c;还能规划决策&#xff0c;扩散强制整合下一token预测与全序列扩散 当前&#xff0c;采用下一 token 预测范式的自回归大型语言模型已经风靡全球&#xff0c;同时互联…...

从零开始深度学习:全连接层、损失函数与梯度下降的详尽指南

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

Liebherr利勃海尔 EDI 需求分析

Liebherr 使用 EDI 技术来提高业务流程的效率、降低错误率、加快数据交换速度&#xff0c;并优化与供应商、客户和其他合作伙伴之间的业务沟通。通过 EDI&#xff0c;Liebherr 实现了与全球交易伙伴的自动化数据交换&#xff0c;提升了供应链管理和订单处理的透明度。 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语言中&#xff0c;占位符主要用于格式化字符串输出&#xff0c;特别是在使用fmt包中的Printf系列函数时。以下是Go语言中常用的占位符&#xff1a; %v&#xff1a;代表值的默认格式&#xff0c;对于字符串是直接输出&#xff0c;对于整型是十进制形式。%v&#xff1a;扩展…...

基于Windows安装opus python库

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

【设计模式】行为型模式(五):解释器模式、访问者模式、依赖注入

《设计模式之行为型模式》系列&#xff0c;共包含以下文章&#xff1a; 行为型模式&#xff08;一&#xff09;&#xff1a;模板方法模式、观察者模式行为型模式&#xff08;二&#xff09;&#xff1a;策略模式、命令模式行为型模式&#xff08;三&#xff09;&#xff1a;责…...

使用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的安装与免密码登录的配置

&#x1f351;个人主页&#xff1a;Jupiter. &#x1f680; 所属专栏&#xff1a;MySQL初阶探索&#xff1a;构建数据库基础 欢迎大家点赞收藏评论&#x1f60a; 目录 &#x1f4da;mysql的安装&#x1f4d5;MySQL的登录&#x1f30f;MySQL配置免密码登录 &#x1f4da;mysql的…...

高级 SQL 技巧讲解

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

浅论AI大模型在电商行业的发展未来

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

【python笔记03】《类》

文章目录 面向对象基本概念对象的概念类的概念 类的定义类的创建&#xff08;实例的模板&#xff09;类的实例化--获取对象对象方法中的self关键字面试题请描述什么是对象&#xff0c;什么是类。请观阅读如下代码&#xff0c;判断是否能正常运行&#xff0c;如果不能正常运行&a…...

Flutter 应用在真机上调试的流程

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

以太坊基础知识结构详解

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

安全见闻(完整版)

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

LeetCode100之反转链表(206)--Java

1.问题描述 给你单链表的头节点 head &#xff0c;请你反转链表&#xff0c;并返回反转后的链表 示例1 输入&#xff1a;head [1,2,3,4,5] 输出&#xff1a;[5,4,3,2,1] 示例2 输入&#xff1a;head [1,2] 输出&#xff1a;[2,1] 示例3 输入&#xff1a;head [] 输…...

牛客周赛第一题2024/11/17日

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

麒麟Server下安装东方通TongLINK/Q

环境 系统&#xff1a;麒麟Server SP3 2403 应用&#xff1a;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 来从百度搜索结果中提取信息。以下是完整的代码&#xff0c;包括项目结构、README.md 文件以及所有必要的代码。 项目结构 xihe241117/ ├── data/ │ └── train_data.jsonl ├── lo…...

Ubuntu下的Eigen库的安装及基本使用教程

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

【spring 】Spring Cloud Gateway 的Filter学习

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

每秒交易数(Transactions Per Second:TPS)详细拆解

每秒交易数&#xff08;TPS&#xff09;是指计算机网络每秒可以处理的交易数量。TPS是衡量不同区块链和其他计算机系统速度的关键指标。然而&#xff0c;TPS并不是用来衡量区块链速度的唯一指标。许多人认为&#xff0c;虽然TPS很重要&#xff0c;但最终性实际上是一个更重要的…...