Flink_DataStreamAPI_执行环境

DataStreamAPI_执行环境

  • 1创建执行环境
    • 1.1getExecutionEnvironment
    • 1.2createLocalEnvironment
    • 1.3createRemoteEnvironment
  • 2执行模式(Execution Mode)
  • 3触发程序执行

Flink程序可以在各种上下文环境中运行:我们可以在本地JVM中执行程序,也可以提交到远程集群上运行。不同的环境,代码的提交运行的过程会有所不同。这就要求我们在提交作业执行计算时,首先必须获取当前Flink的运行环境,从而建立起与Flink框架之间的联系。

在这里插入图片描述

1创建执行环境

我们要获取的执行环境,是StreamExecutionEnvironment类的对象,这是所有Flink程序的基础。在代码中创建执行环境的方式,就是调用这个类的静态方法,具体有以下三种。

1.1getExecutionEnvironment

最简单的方式,就是直接调用getExecutionEnvironment方法。它会根据当前运行的上下文直接得到正确的结果:如果程序是独立运行的,就返回一个本地执行环境;如果是创建了jar包,然后从命令行调用它并提交到集群执行,那么就返回集群的执行环境。也就是说,这个方法会根据当前运行的方式,自行决定该返回什么样的运行环境。这种方式,用起来简单高效,是最常用的一种创建执行环境的方式。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

1.2createLocalEnvironment

这个方法返回一个本地执行环境。可以在调用时传入一个参数,指定默认的并行度;如果不传入,则默认并行度就是本地的CPU核心数。

StreamExecutionEnvironment localEnv = StreamExecutionEnvironment.createLocalEnvironment();

1.3createRemoteEnvironment

这个方法返回集群执行环境。需要在调用时指定JobManager的主机名和端口号,并指定要在集群中运行的Jar包。在获取到程序执行环境后,我们还可以对执行环境进行灵活的设置。比如可以全局设置程序的并行度、禁用算子链,还可以定义程序的时间语义、配置容错机制。

StreamExecutionEnvironment remoteEnv = StreamExecutionEnvironment.createRemoteEnvironment("host",                   // JobManager主机名1234,                     // JobManager进程端口号"path/to/jarFile.jar"  // 提交给JobManager的JAR包);

2执行模式(Execution Mode)

从Flink 1.12开始,官方推荐的做法是直接使用DataStream API,在提交任务时通过将执行模式设为BATCH来进行批处理。不建议使用DataSet API。

// 流处理环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream API执行模式包括:流执行模式、批执行模式和自动模式。

流执行模式(Streaming)
这是DataStream API最经典的模式,一般用于需要持续实时处理的无界数据流。默认情况下,程序使用的就是Streaming执行模式。

批执行模式(Batch)
专门用于批处理的执行模式。

自动模式(AutoMatic)
在这种模式下,将由程序根据输入数据源是否有界,来自动选择执行模式。
批执行模式的使用。主要有两种方式:
(1)通过命令行配置
在提交作业时,增加execution.runtime-mode参数,指定值为BATCH。

bin/flink run -Dexecution.runtime-mode=BATCH ...

(2)通过代码配置
在代码中,直接基于执行环境调用setRuntimeMode方法,传入BATCH模式。
实际应用中一般不会在代码中配置,而是使用命令行,这样更加灵活。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);

3触发程序执行

需要注意的是,写完输出(sink)操作并不代表程序已经结束。因为当main()方法被调用时,其实只是定义了作业的每个执行操作,然后添加到数据流图中;这时并没有真正处理数据——因为数据可能还没来。Flink是由事件驱动的,只有等到数据到来,才会触发真正的计算,这也被称为“延迟执行”或“懒执行”。
所以我们需要显式地调用执行环境的execute()方法,来触发程序执行。execute()方法将一直等待作业完成,然后返回一个执行结果(JobExecutionResult)。

env.execute();

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/bicheng/60680.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

数据结构——排序(续集)

♥♥♥~~~~~~欢迎光临知星小度博客空间~~~~~~♥♥♥ ♥♥♥零星地变得优秀~也能拼凑出星河~♥♥♥ ♥♥♥我们一起努力成为更好的自己~♥♥♥ ♥♥♥如果这一篇博客对你有帮助~别忘了点赞分享哦~♥♥♥ ♥♥♥如果有什么问题可以评论区留言或者私信我哦~♥♥♥ ✨✨✨✨✨✨ 个…

前端js用canvas合成图片并转file对象

强大的前端js可以做很多东西&#xff0c;今天用canvas合成图片并添加文字&#xff0c;然后转成file对象进行上传。代码如下: <!DOCTYPE html> <html> <head><meta charset"UTF-8"> </head> <body><canvas id"myCanvas&…

C#里怎么样判断一个数是偶数还是奇数

一般是采用取余的做法。 程序如下&#xff1a; /** C# Program to Check whether the Entered Number is Even or Odd*/ using System; using System.Collections.Generic; using System.Linq; using System.Text;namespace check1 {class Program{static void Main(string[]…

linux文本管理!!!

文章目录 第1章 文本过滤/查看命令1.echo&#xff1a;输出文本2.cat&#xff1a;合并文件或查看文件内容3.head&#xff1a;显示文件头部信息4.tail&#xff1a;显示文件尾部信息5.wc: 统计文本行号6.less&#xff1a;分页显示文件内容7.grep&#xff1a;文本过滤工具8.定向符号…

【网络】深入理解 HTTPS:确保数据传输安全的核心协议

目录 引言一、HTTPS的基本概念1.1 什么是 HTTPS&#xff1f;1.2 HTTPS 的工作原理1.3 图解&#xff1a;HTTPS 通信过程1.4 HTTPS 与 HTTP 的区别1.5 为什么 HTTPS 更加重要&#xff1f; 二、SSL/TLS协议的核心2.1 SSL/TLS 协议的作用2.2 SSL/TLS 的工作流程2.2.1 握手阶段2.2.2…

MongoDB在现代Web开发中的应用

&#x1f493; 博客主页&#xff1a;瑕疵的CSDN主页 &#x1f4dd; Gitee主页&#xff1a;瑕疵的gitee主页 ⏩ 文章专栏&#xff1a;《热点资讯》 MongoDB在现代Web开发中的应用 MongoDB在现代Web开发中的应用 MongoDB在现代Web开发中的应用 引言 MongoDB 概述 定义与原理 发展…

CSS的综合应用例子(网页制作)

这是html的一些最基本内容的代码&#xff1a; <!DOCTYPE html> <html lang"zh"> <head> <meta charset"UTF-8"> <meta name"viewport" content"widthdevice-width, initial-scale1.0"> <t…

【设计模式】行为型模式(二):策略模式、命令模式

行为型模式&#xff08;二&#xff09;&#xff1a;策略模式、命令模式 3.策略模式&#xff08;Strategy&#xff09;3.1 示例3.1.1 定义策略接口3.1.2 实现具体策略3.1.3 定义上下文类3.1.4 客户端代码3.1.5 输出结果 3.2 总结3.2.1 优点3.2.2 缺点 4.命令模式&#xff08;Com…

rocketmq——docker-compose安装

rocketmq安装 创建文件夹&#xff0c;这里我们分别部署namesrv和broker 1、namesrv.conf listenPort98762、broker.conf # 所属集群名字 brokerClusterNameDefaultCluster # broker 名字&#xff0c;注意此处不同的配置文件填写的不一样&#xff0c;如果在 broker-a.propert…

OpenGL ES 文字渲染方式有几种?

在音视频或 OpenGL 开发中,文字渲染是一个高频使用的功能,比如制作一些酷炫的字幕、为视频添加水印、设置特殊字体等等。 实际上 OpenGL 并没有定义渲染文字的方式,所以我们最能想到的办法是:将带有文字的图像上传到纹理,然后进行纹理贴图。 本文分别介绍下在应用层和 C+…

飞牛云fnOS本地部署WordPress个人网站并一键发布公网远程访问

文章目录 前言1. Docker下载源设置2. Docker下载WordPress3. Docker部署Mysql数据库4. WordPress 参数设置5. 飞牛云安装Cpolar工具6. 固定Cpolar公网地址7. 修改WordPress配置文件8. 公网域名访问WordPress 前言 本文旨在详细介绍如何在飞牛云NAS上利用Docker部署WordPress&a…

字符串处理指南:Air780E软件的全新视角

今天我会把 Air780E软件的字符串处理详细解析&#xff0c;指南如下&#xff1a; 1、Lua字符串介绍 关于字符串&#xff0c;Lua提供了一些灵活且强大的功能&#xff0c;一些入门知识如下&#xff1a; 1.1 字符串定义 在Lua中&#xff0c;字符串可以用单引号或双引号"来定…

Oracle 聚集因子factor clustering

文章目录 聚集因子(Factor clustering)举例说明查询聚集因子聚集因子的优化结论 最近发现突然忘记聚集因子的原理了&#xff0c;故整理记录一下 聚集因子(Factor clustering) 在Oracle中&#xff0c;聚集因子&#xff08;Clustering Factor&#xff09;用于衡量数据在表中存储…

酒店行业数据仓库

重要名词&#xff1a; PMS&#xff1a;酒店管理系统CRS&#xff1a;中央预定系统客户&#xff1a;可以分为会员、散客&#xff08;自行到店入住&#xff09;、协议&#xff08;与酒店长期合作&#xff0c;内部价&#xff09;、中介预定&#xff1a;可以分为线上预定、线下预定…

C/C++语言 多项式加法和乘法

多项式加法和乘法 多项式的加法题目描述输入输出样例步骤代码段全局变量设定新建结点合并链表 完整代码 多项式乘法题目描述输入输出样例代码段计算两多项式结果输入 完整代码 多项式的加法 题目描述 输入输出 样例 步骤 总体思想是用链表来做 ① 我们发现输入样例中&#xf…

手动实现promise的all,race,finally方法

Promise.all 是一个非常有用的工具&#xff0c;它接受一个 Promise 对象数组&#xff0c;并返回一个新的 Promise。当所有输入的 Promise 都成功解决时&#xff0c;新的 Promise 会解决为一个包含所有结果的数组&#xff1b;如果任何一个 Promise 被拒绝&#xff0c;新的 Prom…

第十四章 Spring之假如让你来写AOP——雏形篇

Spring源码阅读目录 第一部分——IOC篇 第一章 Spring之最熟悉的陌生人——IOC 第二章 Spring之假如让你来写IOC容器——加载资源篇 第三章 Spring之假如让你来写IOC容器——解析配置文件篇 第四章 Spring之假如让你来写IOC容器——XML配置文件篇 第五章 Spring之假如让你来写…

搜维尔科技:Haption力触觉交互,虚拟机械装配验证

Haption力触觉交互&#xff0c;虚拟机械装配验证 搜维尔科技&#xff1a;Haption力触觉交互&#xff0c;虚拟机械装配验证

微信小程序——实现二维码扫描功能(含代码)

✅作者简介&#xff1a;2022年博客新星 第八。热爱国学的Java后端开发者&#xff0c;修心和技术同步精进。 &#x1f34e;个人主页&#xff1a;Java Fans的博客 &#x1f34a;个人信条&#xff1a;不迁怒&#xff0c;不贰过。小知识&#xff0c;大智慧。 &#x1f49e;当前专栏…

PostgreSQL 页损坏如何修复

PostgreSQL 错误&#xff1a;关系 base/46501/52712 中的块 480 存在无效的页。 当我们在使用 PostgreSQL 数据库的时候&#xff0c;如果服务器发生 CRASH 或者断电等异常情况的时候&#xff0c;有可能会遇到上面的这个报错信息。那么我们如何去修复这个数据呢&#xff0c;以及…