Flink_DataStreamAPI_执行环境

embedded/2024/11/17 23:39:13/

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.ppmy.cn/embedded/138379.html

相关文章

Chapter 2 - 15. Understanding Congestion in Fibre Channel Fabrics

Fibre Channel Frame Format The size of a Fibre Channel frame can be up to 2,148 bytes carrying up to 2,112 bytes of payload or data, as shown in Figure 2-25. 如图 2-25 所示,光纤通道帧的大小可达 2,148 字节,最多可携带 2,112 字节的有效载荷或数据。 Figure 2…

什么是 C++ 中的常量表达式? 有什么用途?

在 C 中,常量表达式(constant expression)是指在编译时就能确定值的表达式。这意味着这些表达式的值在程序运行期间不会改变,并且可以在需要常量值的上下文中使用,例如数组大小、模板参数、枚举值等。 常量表达式的用…

道陟科技EMB产品开发进展与标准设计的建议|2024电动汽车智能底盘大会

11月12日,2024电动汽车智能底盘大会在重庆开幕。会议由中国汽车工程学会主办,电动汽车产业技术创新战略联盟、中国汽车工程学会智能底盘分会、智能绿色车辆与交通全国重点实验室承办。本届大会围绕电动汽车智能底盘相关技术发展与融合,满足高…

webman使用中间件验证指定的控制器及方法[青锐CC]

因为webman官方文档没有中间件验证指定的控制器的例子,写了一个小例子。 webman使用中间件验证指定的控制器及方法很简单,只需要三部: 1、新建一个中间件文件。 2、配置文件添加全局中间件。 3、指定的控制器中设置不需要验证的方法。 一…

小白如何快速上手python爬虫?

对于Python爬虫的小白学习路线,可以按照以下步骤进行: python爬虫课程资料: https://kakatu.top/m?sdata Python基础:首先,需要掌握Python的基本语法,包括变量、数据类型、控制流(if/for/whil…

除了 TON, 哪些公链在争夺 Telegram 用户?数据表现如何?

作者:Stella L (stellafootprint.network) 在 2024 年,区块链游戏大规模采用迎来了一个意想不到的催化剂:Telegram。随着各大公链争相布局这个拥有海量用户基础的即时通讯平台,一个核心问题浮出水面:这种用户获取策略…

docker构建jdk11

# 建立一个新的镜像文件,配置模板:新建立的镜像是以centos为基础模板 # 因为jdk必须运行在操作系统之上 FROM centos:7.9.2009# 作者名 MAINTAINER yuanhang# 创建一个新目录来存储jdk文件 RUN mkdir /usr/local/java#将jdk压缩文件复制到镜像中&#…

MySQL基础(初阶+进阶)(详解)

前言: MySQL基础的操作还有很多,上一次主要是针对MySQL的库的增删查的相关操作,接下来就来探究一下MySQL中表的更多的操作。主要围绕着MySQL中表增删查改的操作进行探究! 表的操作: 在进行对表的操作的时候&#xff0…