【API篇】一、Flink执行环境API

喜欢ヅ旅行 2024-02-22 09:03 11阅读 0赞

#### 文章目录 ####

*  0、认识
 *  1、创建执行环境
 *  2、执行模式
 *  3、触发程序执行
 *  4、关于executeAsync方法

## 0、认识 ##

DataStream API是Flink的核心层API。一个Flink程序，其实就是对数据源DataStream的各种转换。具体来说，代码基本上都由以下几部分构成：

![在这里插入图片描述][3257d3d8c9ca45418676b45a8bed24d0.png]

后面章节，分别对每一个环节的API做整理。

## 1、创建执行环境 ##

Flink的程序可以在各种上下文环境中运行，比如本地JVM中执行，也可以提交到远程集群中运行，分别对应着不同的Flink的运行环境，获取这个执行环境，也就是StreamExecutionEnvironment类的对象。

> 获取方式一：

StreamExecutionEnvironment.getExecutionEnvironment()

调用静态方法getExecutionEnvironment，会根据当前运行的上下文直接得到正确的结果：

*  如果程序是独立运行的，就返回一个本地执行环境
 *  如果是命令行+jar包提交到集群执行，就返回集群的执行环境

重点：这个静态方法根据当前运行方式，自行决定并返回一个适配的运行环境。 getExecutionEnvironment方法还可以传一个flink包下的Confiruration对象

StreamExecutionEnvironment.getExecutionEnvironment(new Configuration());

此时，就可以访问web控制台了。localhost:8081。此外，通过Configuraiton对象，可以改一些默认的配置，比如端口8081

Configuration conf = new Configuration();
    
    conf.set(RestOptions.BIND_PORT, "8082");

此时，控制台就该访问localhost:8082

> 获取方式二：

StreamExecutionEnvironment.createLocalEnvironment()

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

//并行度为3
    StreamExecutionEnvironment localEnv = StreamExecutionEnvironment.createLocalEnvironment(3);

> 获取方式三：

这个方法返回集群执行环境。需要在调用时指定JobManager的主机名和端口号，并指定要在集群中运行的Jar包。

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

最后，不管用哪个方法，拿到执行环境对象后，还可以全局设置程序的并行度、禁用算子链，还可以定义程序的时间语义、配置容错机制等

env.setParallelism(2);

## 2、执行模式 ##

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

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

DataStream API执行模式包括：

*  流执行模式（Streaming）：用于需要持续实时处理的无界数据流。默认情况下，程序使用的就是Streaming执行模式
 *  批执行模式（Batch）：专门用于批处理的执行模式
 *  自动模式（AutoMatic）：由程序根据输入数据源是否有界，来自动选择是流处理还是批处理执行

执行模式选择，可以通过命令行方式配置：

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

也可以通过代码配置：

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

前者用的更多，不推荐硬编码。

## 3、触发程序执行 ##

写完输出（sink）操作并不代表程序已经结束，因为当main()方法被调用时，其实只是定义了作业的每个执行操作，然后添加到数据流图中，这时并没有真正处理数据，因为数据可能还没来。Flink是由事件驱动的，只有等到数据到来，才会触发真正的计算，这也被称为**延迟执行**或**懒执行**。

env.execute();

所以我们需要显式地调用执行环境的execute()方法，来触发程序执行。execute()方法将一直等待作业完成，然后才返回一个执行结果（JobExecutionResult）

## 4、关于executeAsync方法 ##

*  正常情况下，一个execute方法执行，触发一个Flink Job。
 *  一个main方法里也可以调用多个execute，但没意义，因为execute()方法将一直等待作业完成，肯定阻塞后面的
 *  env.executeAsync方法，可以异步触发，不会阻塞后面的任务

public static void main(String[] args){
        
    
    	...
    	executeAsync();   //触发第一个Job
    	
    	//job2...(一般不这么写)
    	
    	....
    	executeAsync();  //第二个Job产生
    }

*  一个main方法里executeAsync方法的调用次数，等于生成的flink job的个数
 *  同样的，再Yarn-Application集群，提交一次，当调用n次executeAsync，就有n个Job，对应在JobManager里，就有n个JobMaster

[3257d3d8c9ca45418676b45a8bed24d0.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/22/5e71a39479ee4488a59b388c4478f935.png