Apache Flink流作业提交流程分析

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: 提交流程调用的关键方法链 用户编写的程序逻辑需要提交给Flink才能得到执行。本文来探讨一下客户程序如何提交给Flink。鉴于用户将自己利用Flink的API编写的逻辑打成相应的应用程序包(比如Jar)然后提交到一个目标Flink集群上去运行是比较主流的使用场景,因此我们的分析也基于这一场景进行。

用户编写的程序逻辑需要提交给Flink才能得到执行。本文来探讨一下客户程序如何提交给Flink。鉴于用户将自己利用Flink的API编写的逻辑打成相应的应用程序包(比如Jar)然后提交到一个目标Flink集群上去运行是比较主流的使用场景,因此我们的分析也基于这一场景进行。

Flink的API针对不同的执行环境有不同的Environment对象,这里我们主要基于常用的RemoteStreamEnvironmentRemoteEnvironment进行分析

在前面我们谈到了Flink中实现了“惰性求值”,只有当最终调用execute方法时,才会“真正”开始执行。因此,execute方法是我们的切入点。

其源码位于org.apache.flink.streaming.api.environment.RemoteStreamEnvironment

首先,我们来看一下其execute方法触发的关键方法调用链示意图:

streaming-remoge-execute-method-chain

根据上图的调用链,我们针对这些关键方法进行剖析,当然一些细节性的内容我们可能会暂时略过,这样可以保证主路径一直都很清晰。

getStreamGraph方法用于获得一个StreamGraph的实例,该实例表示流的完整的拓扑结构并且包含了生成JobGraph所必要的相关信息(包含了sourcesink的集合以及这些在图中的“节点”抽象化的表示、一些虚拟的映射关系、执行和检查点的配置等)。

获得StreamGraph之后,通过调用executeRemotely方法进行远程执行。该方法首先根据获取到的用户程序包的路径以及类路径创建加载用户代码的类加载器:

ClassLoader usercodeClassLoader = JobWithJars.buildUserCodeClassLoader(
jarFiles, globalClasspaths,   getClass().getClassLoader());

紧接着根据配置构建Client对象(Client对象是真正跟JobManager对接的内部代理):

Client client;
try {   
    client = new Client(configuration);   
    client.setPrintStatusDuringExecution(getConfig().isSysoutLoggingEnabled());
}catch (Exception e) {
   throw new ProgramInvocationException("Cannot establish connection to JobManager: " + e.getMessage(), e);
}

后面的事情就此被Client接管:

try {   
    return client.runBlocking(streamGraph, jarFiles, globalClasspaths, usercodeClassLoader);
}catch (ProgramInvocationException e) {   
    throw e;
}catch (Exception e) {   
    String term = e.getMessage() == null ? "." : (": " + e.getMessage());   
    throw new ProgramInvocationException("The program execution failed" + term, e);
}finally {   
    client.shutdown();
}

client对象调用了runBlocking以阻塞式的行为“运行”用户程序并等待返回JobExecutionResult对象作为Job的执行结果。执行完成,最终在finally块中,调用shutdown方法关闭并释放资源。

runBlocking被调用后,调用链跳转到Client类中。为了适配多种提交方式以及运行模式,runBlocking方法有着非常多的重载。在当前的远程执行环境下,runBlocking在多个重载方法之间跳转的过程中,会调用getJobGraph方法获得JobGraph的实例。JobGraph表示Flink dataflow 程序,它将会被JobManager所理解并接收。在某个Job被提交给JobManager之前,通过Flink提供的高层次的API都将会被转化为JobGraph表示。关于如何获得JobGraph的实现,我们后面会进行剖析。这里,让我们忽视这些细节,进入下一个关键方法。

runBlocking_1其实是runBlocking方法的重载,这里加一个后缀标识,只是为了跟上面的runBlocking进行区别。runBlocking_1方法中,首先利用LeaderRetrievalUtils创建了LeaderRetrievalService这一服务对象:

LeaderRetrievalService leaderRetrievalService;
try {   
    leaderRetrievalService = LeaderRetrievalUtils.createLeaderRetrievalService(config);
} catch (Exception e) {   
    throw new ProgramInvocationException("Could not create the leader retrieval service.", e);
}

顾名思义,LeaderRetrievalService在Flink中提供查找主节点的服务。它会根据Flink的配置信息(主要是recovery.mode来判断基于哪种恢复机制来创建该服务。当前有两种模式:一种是Standalone的独立运行模式;另一种是基于Zookeeper的高可用模式)。Flink提供了一个称之为LeaderRetrievalListener的回调接口来获得主节点的信息。接下来,就是调用JobClientsubmitJobAndWait方法将产生的JobGraph以及主节点查找的服务对象等相关信息提交给JobManager并等待返回结果:

try {   
    this.lastJobID = jobGraph.getJobID();   
    return JobClient.submitJobAndWait(actorSystem, leaderRetrievalService, jobGraph, 
                                        timeout, printStatusDuringExecution, classLoader);
} catch (JobExecutionException e) {   
    throw new ProgramInvocationException("The program execution failed: " + e.getMessage(), e);
}

上面的submitJobAndWait方法的第一个参数actorSystemActorSystem的实例。在构造Client对象时创建,在Job提交并获得返回结果后通过调用Clientshutdown方法关闭:

public void shutdown() {   
    if (!this.actorSystem.isTerminated()) {      
        this.actorSystem.shutdown();      
        this.actorSystem.awaitTermination();   
    }
}

该方法的调用见上面executeRemotely方法的代码段的finally语句块。

JobClient的出现可能会让你产生疑惑——它跟Client是什么关系?作用是什么?下面这幅示意图可以用来解释这些疑问:

Client-JobClient-relationship

上面这幅图展示了Client对象与其他几个对象的关系。JobClient在其中起到了“桥接”作用,它在基于API的编程层面上桥接了同步的方法调用和异步的消息通信。更具体得说,JobClient可以看做是一个“静态类”提供了一些静态方法,这里我们主要关注上面的submitJobAndWait方法,该方法内部封装了Actor之间的异步通信(具体的通信对象是JobClientActor,它负责跟JobManagerActorSystemActor进行通信),并以阻塞的形式返回结果。而Client只需调用JobClient的这些方法,而无需关注其内部是如何实现的。

通过调用JobClient的静态方法submitJobAndWait,会触发基于AkkaActor之间的消息通信来完成后续的提交JobGraph的动作。JobClient提交Job的基于消息交互的抽象示意图如下:

JobClient-Actor-SubmitJobGraph

总体来说这里总共有两个ActorSystem,一个归属于Client,另一个归属于JobManager。在submitJobAndWait方法中,其首先会创建一个JobClientActorActorRef

ActorRef jobClientActor = actorSystem.actorOf(jobClientActorProps);

然后向其发起一个SubmitJobAndWait消息,该消息将JobGraph的实例提交给jobClientActor。该消息的发起模式是ask,它表示需要一个应答消息。

JobClient向JobClientActor发送消息的代码段如下所示:

Future<Object> future = Patterns.ask(jobClientActor,      
                                     new JobClientMessages.SubmitJobAndWait(jobGraph),      
                                     new Timeout(AkkaUtils.INF_TIMEOUT()));
answer = Await.result(future, AkkaUtils.INF_TIMEOUT());

JobClient会阻塞等待该future返回结果。在得到返回结果answer之后,先进行解析判断它是Job被成功执行返回的结果还是失败返回的结果。

小结

至此,Client提交Streaming Job的关键方法调用路径已梳理完成。这里为了突出主路线,同时避免被太多的实现细节干扰,我们暂时忽略了一些重要数据结构和关键概念的解读。不过,后续我们会对它们进行分析。




原文发布时间为:2016-07-17


本文作者:vinoYang


本文来自云栖社区合作伙伴CSDN博客,了解相关信息可以关注CSDN博客。

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
目录
相关文章
|
5天前
|
自然语言处理 监控 数据挖掘
【Flink】Flink中的窗口分析
【4月更文挑战第19天】【Flink】Flink中的窗口分析
|
5天前
|
存储 Apache
Apache Hudi Savepoint实现分析
Apache Hudi Savepoint实现分析
42 0
|
5天前
|
Apache
Apache Hudi Rollback实现分析
Apache Hudi Rollback实现分析
27 0
|
1天前
|
Oracle 关系型数据库 数据库
实时计算 Flink版操作报错合集之执行Flink job,报错“Could not execute SQL statement. Reason:org.apache.flink.table.api.ValidationException: One or more required options are missing”,该怎么办
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
9 0
|
2天前
|
消息中间件 关系型数据库 MySQL
实时计算 Flink版操作报错合集之遇到报错:Apache Kafka Connect错误如何解决
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
13 5
|
2天前
|
SQL 关系型数据库 MySQL
实时计算 Flink版操作报错合集之报错:org.apache.flink.table.api.validationexception如何解决
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
11 1
|
2天前
|
存储 SQL 关系型数据库
实时计算 Flink版操作报错合集之报错:WARN (org.apache.kafka.clients.consumer.ConsumerConfig:logUnused)这个错误如何解决
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
13 3
|
5天前
|
消息中间件 API Apache
官宣|阿里巴巴捐赠的 Flink CDC 项目正式加入 Apache 基金会
本文整理自阿里云开源大数据平台徐榜江 (雪尽),关于阿里巴巴捐赠的 Flink CDC 项目正式加入 Apache 基金会。
1810 2
官宣|阿里巴巴捐赠的 Flink CDC 项目正式加入 Apache 基金会
|
5天前
|
SQL Java API
官宣|Apache Flink 1.19 发布公告
Apache Flink PMC(项目管理委员)很高兴地宣布发布 Apache Flink 1.19.0。
1797 2
官宣|Apache Flink 1.19 发布公告
|
5天前
|
SQL 缓存 分布式计算
flink1.18 SqlGateway 的使用和原理分析
# 了解flink1.18 sqlGateway 的安装和使用步骤 # 启动sqlgateway 流程,了解核心的结构 # sql提交流程,了解sql 的流转逻辑 # select 查询的ResultSet的对接流程,了解数据的返回和获取逻辑

推荐镜像

更多