Flink 1.9源码学习04 ----Job启动Driver端

素颜马尾好姑娘i 2023-06-06 07:49 1阅读 0赞

1，整个Flink的Job启动是通过在Driver端通过用户的Envirement的execute()方法将用户的算子转化成StreamGraph

2，然后得到JobGraph通过远程RPC将这个JobGraph提交到JobManager对应的接口

3，JobManager转化成executionGraph.deploy()，然后生成TDD发给TaskManager，然后整个Job就启动起来了

这里来看一下Driver端的实现从用户的Envirement.execute()方法作为入口：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]

\--->>>

继续点进去方法：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1][]

\--->>>

继续点进去：

>     private static JobExecutionResult executeRemotely(StreamGraph streamGraph,
>        ClassLoader envClassLoader,
>        ExecutionConfig executionConfig,
>        List<URL> jarFiles,
>        String host,
>        int port,
>        Configuration clientConfiguration,
>        List<URL> globalClasspaths,
>        SavepointRestoreSettings savepointRestoreSettings
>     ) throws ProgramInvocationException {
>        if (LOG.isInfoEnabled()) {
>           LOG.info("Running remotely at {}:{}", host, port);
>        }
>     
>        ClassLoader userCodeClassLoader = JobWithJars.buildUserCodeClassLoader(jarFiles, globalClasspaths, envClassLoader);
>     
>        Configuration configuration = new Configuration();
>        configuration.addAll(clientConfiguration);
>     
>        configuration.setString(JobManagerOptions.ADDRESS, host);
>        configuration.setInteger(JobManagerOptions.PORT, port);
>     
>        configuration.setInteger(RestOptions.PORT, port);
>     
>        final ClusterClient<?> client;
>        try {
>           client = new RestClusterClient<>(configuration, "RemoteStreamEnvironment");
>        }
>        catch (Exception e) {
>           throw new ProgramInvocationException("Cannot establish connection to JobManager: " + e.getMessage(),
>              streamGraph.getJobGraph().getJobID(), e);
>        }
>     
>        client.setPrintStatusDuringExecution(executionConfig.isSysoutLoggingEnabled());
>     
>        if (savepointRestoreSettings == null) {
>           savepointRestoreSettings = SavepointRestoreSettings.none();
>        }
>     
>        try {
>           //todo 返回的JobExecutionResult对象
>           return client.run(streamGraph, jarFiles, globalClasspaths, userCodeClassLoader, savepointRestoreSettings)
>              .getJobExecutionResult();
>        }
>        catch (ProgramInvocationException e) {
>           throw e;
>        }
>        catch (Exception e) {
>           String term = e.getMessage() == null ? "." : (": " + e.getMessage());
>           throw new ProgramInvocationException("The program execution failed" + term,
>              streamGraph.getJobGraph().getJobID(), e);
>        }
>        finally {
>           try {
>              client.shutdown();
>           } catch (Exception e) {
>              LOG.warn("Could not properly shut down the cluster client.", e);
>           }
>        }
>     }

\--->>>

进入client.run()

在这里我们发现了，StreamGraph  转成了 JobGraph

![20191010143601809.png][]

通过查看类StreamGraph  发现他是继承StreamingPlan

package org.apache.flink.streaming.api.graph.StreamGraph;

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 3][]

\--->>>

转换的具体方法：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 4][]

\--->>>

再回到上级，我们找到提交job的具体实现：

>     @Override
>     public JobSubmissionResult submitJob(JobGraph jobGraph, ClassLoader classLoader) throws ProgramInvocationException {
>        final CompletableFuture<JobSubmissionResult> jobSubmissionResultFuture = submitJob(jobGraph);
>     
>        if (isDetached()) {
>           try {
>              return jobSubmissionResultFuture.get();
>           } catch (InterruptedException | ExecutionException e) {
>              ExceptionUtils.checkInterrupted(e);
>     
>              throw new ProgramInvocationException("Could not run job in detached mode.", jobGraph.getJobID(), e);
>           }
>        } else {
>           final CompletableFuture<JobResult> jobResultFuture = jobSubmissionResultFuture.thenCompose(
>              (JobSubmissionResult ignored) -> requestJobResult(jobGraph.getJobID()));
>     
>           final JobResult jobResult;
>           try {
>              jobResult = jobResultFuture.get();
>           } catch (InterruptedException | ExecutionException e) {
>              ExceptionUtils.checkInterrupted(e);
>     
>              throw new ProgramInvocationException("Could not run job", jobGraph.getJobID(), e);
>           }
>     
>           try {
>              return jobResult.toJobExecutionResult(classLoader);
>           } catch (JobExecutionException e) {
>              throw new ProgramInvocationException("Job failed", jobGraph.getJobID(), e);
>           } catch (IOException | ClassNotFoundException e) {
>              throw new ProgramInvocationException("Job failed", jobGraph.getJobID(), e);
>           }
>        }
>     }

具体的执行toJobExecutionResult（）方法：

看到抛出的异常提示，是不是很熟悉

>     public JobExecutionResult toJobExecutionResult(ClassLoader classLoader) throws JobExecutionException, IOException, ClassNotFoundException {
>        if (applicationStatus == ApplicationStatus.SUCCEEDED) {
>           return new JobExecutionResult(
>              jobId,
>              netRuntime,
>              AccumulatorHelper.deserializeAccumulators(
>                 accumulatorResults,
>                 classLoader));
>        } else {
>           final Throwable cause;
>     
>           if (serializedThrowable == null) {
>              cause = null;
>           } else {
>              cause = serializedThrowable.deserializeError(classLoader);
>           }
>     
>           final JobExecutionException exception;
>     
>           if (applicationStatus == ApplicationStatus.FAILED) {
>              exception = new JobExecutionException(jobId, "Job execution failed.", cause);
>           } else if (applicationStatus == ApplicationStatus.CANCELED) {
>              exception = new JobCancellationException(jobId, "Job was cancelled.", cause);
>           } else {
>              exception = new JobExecutionException(jobId, "Job completed with illegal application status: " + applicationStatus + '.', cause);
>           }
>     
>           throw exception;
>        }
>     }

顺便看一下JobExecutionResult这个类的组成

主要是jobid ，作业运行时间，一个map类型的 accumulators

：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 5][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: /images/20230601/2f801bcc835548078b22bc4f191e40fa.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1]: /images/20230601/5c04abe910a245549d6b9cfbf5ccd2d0.png
[20191010143601809.png]: /images/20230601/c2df213f569d46c39f1f86c390590d51.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 2]: /images/20230601/e2d52dc23d3f4ee787a7dca5eff3a987.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 3]: /images/20230601/4a9b432e003e4daeade35e2745eb2159.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 4]: /images/20230601/eb91632fe1ce4e44b23c8e86805112a8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 5]: /images/20230601/87a47155c9f54ffbac63603525640960.png