Flink_DataStreamAPI_实行情况
Flink程序可以在各种上下文情况中运行：我们可以在本地JVM中实行程序，也可以提交到远程集群上运行。不同的情况，代码的提交运行的过程会有所不同。这就要求我们在提交作业实行盘算时，首先必须获取当前Flink的运行情况，从而建立起与Flink框架之间的联系。https://i-blog.csdnimg.cn/direct/0f78db6791e94d21b34559f672d07e5c.png
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();
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]