Flink Job 执行流程

Flink On Yarn 模式

​
基于Yarn层面的架构雷同 Spark on Yarn模式，都是由Client提交App到RM上面去运行，然后 RM分配第一个container去运行AM，然后由AM去负责资源的监督和管理。必要分析的是，Flink的Yarn模式更加雷同Spark on Yarn的cluster模式，在cluster模式中，dirver将作为AM中的一个线程去运行。Flink on Yarn模式也是会将JobManager启动在container里面，去做个driver雷同的使命调度和分配，Yarn AM与Flink JobManager在同一个Container中，如许AM可以知道Flink JobManager的地点，从而AM可以申请Container去启动Flink TaskManager。待Flink乐成运行在Yarn集群上，Flink Yarn Client就可以提交Flink Job到Flink JobManager，并进行后续的映射、调度和计算处置惩罚。
Fink on Yarn 的缺陷

【1】资源分配是静态的，一个作业必要在启动时获取所需的资源而且在它的生命周期里一直持有这些资源。这导致了作业不能随负载变化而动态调解，在负载下降时无法归还空闲的资源，在负载上升时也无法动态扩展。
【2】On-Yarn模式下，所有的container都是固定大小的，导致无法根据作业需求来调解container的布局。譬如CPU密集的作业或必要更多的核，但不必要太多内存，固定布局的container会导致内存被浪费。
【3】与容器管理基础设施的交互比较笨拙，必要两个步骤来启动Flink作业：1.启动Flink守护历程；2.提交作业。如果作业被容器化而且将作业部署作为容器部署的一部分，那么将不再必要步骤2。
【4】On-Yarn模式下，作业管理页面会在作业完成后消失不可访问。
【5】Flink保举 per job clusters 的部署方式，但是又支持可以在一个集群上运行多个作业的session模式，令人迷惑。
在Flink版本1.5中引入了Dispatcher，Dispatcher是在新筹划里引入的一个新概念。Dispatcher会从Client端担当作业提交哀求并代表它在集群管理器上启动作业。引入Dispatcher的原因重要有两点：
【1】一些集群管理器必要一个中心化的作业生成和监控实例；
【2】能够实现Standalone模式下JobManager的角色，且等候作业提交。在一些案例中，Dispatcher是可选的Yarn或者不兼容的kubernetes。
资源调度模子重构下的 Flink On Yarn 模式

客户端提交JobGraph以及依赖jar包到YarnResourceManager，接着Yarn ResourceManager分配第一个container以此来启动AppMaster，Application Master中会启动一个FlinkResourceManager以及JobManager，JobManager会根据JobGraph生成的ExecutionGraph以及物理执行筹划向FlinkResourceManager申请slot，FlinkResoourceManager会管理这些slot以及哀求，如果没有可用slot就向Yarn的ResourceManager申请container，container启动以后会注册到FlinkResourceManager，末了JobManager会将subTask deploy到对应container的 slot中去。

在有Dispatcher的模式下：会增长一个过程，就是Client会直接通过HTTP Server的方式，然后用Dispatcher将这个使命提交到Yarn ResourceManager中。
新框架具有四大优势，详情如下：
【1】client直接在Yarn上启动作业，而不必要先启动一个集群然后再提交作业到集群。因此client再提交作业后可以马上返回。
【2】所有的用户依赖库和配置文件都被直接放在应用的classpath，而不是用动态的用户代码classloader去加载。
【3】container在必要时才哀求，不再使用时会被开释。
【4】“必要时申请”的container分配方式允许差别算子使用差别profile (CPU和内存布局)的container。
新的资源调度框架下 single cluster job on Yarn 流程先容

single cluster job on Yarn模式涉及三个实例对象：
【1】clifrontend： Invoke App code；生成StreamGraph，然后转化为JobGraph；
【2】YarnJobClusterEntrypoint（Master）： 依次启动YarnResourceManager、MinDispatcher、JobManagerRunner三者都服从分布式协同同等的策略；JobManagerRunner将JobGraph转化为ExecutionGraph，然后转化为物理执行使命Execution，然后进行deploy，deploy过程会向 YarnResourceManager哀求slot，如果有直接deploy到对应的YarnTaskExecutiontor的slot里面，没有则向Yarn的ResourceManager申请，带container启动以后deploy。
【3】YarnTaskExecutorRunner (slave)： 负责吸收subTask，并运行。
整个使命运行代码调用流程如下图

subTask在执行时是怎么运行的？
调用StreamTask的invoke方法，执行步骤如下：
【1】initializeState()即operator的initializeState()；
【2】openAllOperators()即operator的open()方法；
【3】末了调用run方法来进行真正的使命处置惩罚；
我们来看下flatMap对应的OneInputStreamTask的run方法具体是怎么处置惩罚的。

1. @Override
2. protected void run() throws Exception {
3.     // 在堆栈上缓存处理器引用，使代码更易于JIT
4.     final StreamInputProcessor<IN> inputProcessor = this.inputProcessor;
6.     while (running && inputProcessor.processInput()) {
7.         // 所有的工作都发生在“processInput”方法中
8.     }
9. }

复制代码

最终是调用StreamInputProcessor的processInput()做数据的处置惩罚，这里面包含用户的处置惩罚逻辑。

1. public boolean processInput() throws Exception {
2.     if (isFinished) {
3.         return false;
4.     }
5.     if (numRecordsIn == null) {
6.         try {
7.             numRecordsIn = ((OperatorMetricGroup) streamOperator.getMetricGroup()).getIOMetricGroup().getNumRecordsInCounter();
8.         } catch (Exception e) {
9.             LOG.warn("An exception occurred during the metrics setup.", e);
10.            numRecordsIn = new SimpleCounter();
11.        }
12.    }
13.    while (true) {
14.        if (currentRecordDeserializer != null) {
15.            DeserializationResult result = currentRecordDeserializer.getNextRecord(deserializationDelegate);
16.            if (result.isBufferConsumed()) {
17.                currentRecordDeserializer.getCurrentBuffer().recycleBuffer();
18.                currentRecordDeserializer = null;
19.            }
20.            if (result.isFullRecord()) {
21.                StreamElement recordOrMark = deserializationDelegate.getInstance();
22.                //处理watermark
23.                if (recordOrMark.isWatermark()) {
24.                    // handle watermark
25.                    //watermark处理逻辑，这里可能引起timer的trigger
26.                    statusWatermarkValve.inputWatermark(recordOrMark.asWatermark(), currentChannel);
27.                    continue;
28.                } else if (recordOrMark.isStreamStatus()) {
29.                    // handle stream status
30.                    statusWatermarkValve.inputStreamStatus(recordOrMark.asStreamStatus(), currentChannel);
31.                    continue;
32.                    //处理latency watermark
33.                } else if (recordOrMark.isLatencyMarker()) {
34.                    // handle latency marker
35.                    synchronized (lock) {
36.                        streamOperator.processLatencyMarker(recordOrMark.asLatencyMarker());
37.                    }
38.                    continue;
39.                } else {
40.                    //用户的真正的代码逻辑
41.                    // now we can do the actual processing
42.                    StreamRecord<IN> record = recordOrMark.asRecord();
43.                    synchronized (lock) {
44.                        numRecordsIn.inc();
45.                        streamOperator.setKeyContextElement1(record);
46.                        //处理数据
47.                        streamOperator.processElement(record);
48.                    }
49.                    return true;
50.                }
51.            }
52.        }
53.             
54.        //这里会进行checkpoint barrier的判断和对齐，以及不同partition 里面checkpoint barrier不一致时候的，数据buffer，
55.        final BufferOrEvent bufferOrEvent = barrierHandler.getNextNonBlocked();
56.        if (bufferOrEvent != null) {
57.            if (bufferOrEvent.isBuffer()) {
58.                currentChannel = bufferOrEvent.getChannelIndex();
59.                currentRecordDeserializer = recordDeserializers[currentChannel];
60.                currentRecordDeserializer.setNextBuffer(bufferOrEvent.getBuffer());
61.            }
62.            else {
63.                // Event received
64.                final AbstractEvent event = bufferOrEvent.getEvent();
65.                if (event.getClass() != EndOfPartitionEvent.class) {
66.                    throw new IOException("Unexpected event: " + event);
67.                }
68.            }
69.        }
70.        else {
71.            isFinished = true;
72.            if (!barrierHandler.isEmpty()) {
73.                throw new IllegalStateException("Trailing data in checkpoint barrier handler.");
74.            }
75.            return false;
76.        }
77.    }
78. }

复制代码

streamOperator.processElement(record)最终会调用用户的代码处置惩罚逻辑，假如operator是StreamFlatMap的话。

1. @Override
2. public void processElement(StreamRecord<IN> element) throws Exception {
3.     collector.setTimestamp(element);
4.     userFunction.flatMap(element.getValue(), collector);//用户代码
5. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。