Spark in action on Kubernetes - Spark Operator的原理解析

本文涉及的产品
应用实时监控服务-应用监控,每月50GB免费额度
应用实时监控服务-用户体验监控,每月100OCU免费额度
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
简介: 作者 | 阿里云智能事业群技术专家 莫源[](https://www.atatech.org/articles/136659#1)Spark Operator的内部实现在深入解析Spark Operator之前,我们先补充一些关于kubernetes operator的知识。

作者 | 阿里云智能事业群技术专家 莫源

[](https://www.atatech.org/articles/136659#1)Spark Operator的内部实现

在深入解析Spark Operator之前,我们先补充一些关于kubernetes operator的知识。2018年可以说是kubernetes operator泛滥的一年,各种operator如雨后春笋般出现。operator是扩展kubernetes以及与kubernetes集成的最佳方式之一。在kubernetes的设计理念中,有很重要的一条就是进行了抽象,比如对存储进行抽象、对应用负载进行抽象、对接入层进行抽象等等。每个抽象又对应了各自生命周期管理的controller,开发者提交的Yaml实际上是对抽象终态的描述,而controller会监听抽象的变化、解析并进行处理,最终尝试将状态修正到终态。

那么对于在kubernetes中未定义的抽象该如何处理呢,答案就是operator。一个标准operator通常包含如下几个部分:1. CRD抽象的定义,负责描述抽象所能包含的功能。 2.CRD Controller ,负责解析CRD定义的内容以及生命周期的管理。3.clent-go的SDK,负责提供代码集成时使用的SDK。

有了这个知识储备,那么我们回过头来看Spark Operator的代码,结构基本就比较明晰了。核心的代码逻辑都在pkg下,其中apis下面主要是定义了不同版本的API;client目录下主要是自动生成的client-go的SDK;crd目录下主要是定义的两个自定义资源sparkapplication和scheduledsparkapplication的结构。controller目录下主要定义的就是这个operator的生命周期管理的逻辑;config目录下主要处理spark config的转换。了解一个Operator能力最快捷的方式,就是查看CRD的定义。在Spark Operator中定义了sparkapplication和scheduledsparkapplication两个CRD,他们之间有什么区别呢?
sparkapplication 是对常规spark任务的抽象,作业是单次运行的,作业运行完毕后,所有的Pod会进入Succeed或者Failed的状态。而scheduledsparkapplication是对离线定时任务的一种抽象,开发者可以在scheduledsparkapplication中定义类似crontab的任务,实现spark离线任务的周期性定时调度。

上面这张图是Spark中kubernetes的集成图,也就是说当我们通过spark-submit提交作业的时候,会自动生成driver pod与exector pods。那么引入了Spark Operator后,这个流程变成了什么呢?

func (c *Controller) submitSparkApplication(app *v1beta1.SparkApplication) *v1beta1.SparkApplication {
    // prometheus的监控指标的暴露
    appToSubmit := app.DeepCopy()
    if appToSubmit.Spec.Monitoring != nil && appToSubmit.Spec.Monitoring.Prometheus != nil {
        if err := configPrometheusMonitoring(appToSubmit, c.kubeClient); err != nil {
            glog.Error(err)
        }
    }
  // 将CRD中的定义转变为spark-submit的命令
    submissionCmdArgs, err := buildSubmissionCommandArgs(appToSubmit)
    if err != nil {
        app.Status = v1beta1.SparkApplicationStatus{
            AppState: v1beta1.ApplicationState{
                State:        v1beta1.FailedSubmissionState,
                ErrorMessage: err.Error(),
            },
            SubmissionAttempts:        app.Status.SubmissionAttempts + 1,
            LastSubmissionAttemptTime: metav1.Now(),
        }
        return app
    }
    // 在operator容器内通过spark-submit提交作业
    submitted, err := runSparkSubmit(newSubmission(submissionCmdArgs, appToSubmit))
    if err != nil {
        app.Status = v1beta1.SparkApplicationStatus{
            AppState: v1beta1.ApplicationState{
                State:        v1beta1.FailedSubmissionState,
                ErrorMessage: err.Error(),
            },
            SubmissionAttempts:        app.Status.SubmissionAttempts + 1,
            LastSubmissionAttemptTime: metav1.Now(),
        }
        c.recordSparkApplicationEvent(app)
        glog.Errorf("failed to run spark-submit for SparkApplication %s/%s: %v", app.Namespace, app.Name, err)
        return app
    }
    // 因为Pod的状态也会被Spark Operator进行观测,因此driver pod宕掉会被重新拉起
    // 这是和直接跑spark-submit的一大区别,提供了故障恢复的能力。
    if !submitted {
        // The application may not have been submitted even if err == nil, e.g., when some
        // state update caused an attempt to re-submit the application, in which case no
        // error gets returned from runSparkSubmit. If this is the case, we simply return.
        return app
    }
    glog.Infof("SparkApplication %s/%s has been submitted", app.Namespace, app.Name)
    app.Status = v1beta1.SparkApplicationStatus{
        AppState: v1beta1.ApplicationState{
            State: v1beta1.SubmittedState,
        },
        SubmissionAttempts:        app.Status.SubmissionAttempts + 1,
        ExecutionAttempts:         app.Status.ExecutionAttempts + 1,
        LastSubmissionAttemptTime: metav1.Now(),
    }
    c.recordSparkApplicationEvent(app)
    // 通过service暴露spark-ui
    service, err := createSparkUIService(app, c.kubeClient)
    if err != nil {
        glog.Errorf("failed to create UI service for SparkApplication %s/%s: %v", app.Namespace, app.Name, err)
    } else {
        app.Status.DriverInfo.WebUIServiceName = service.serviceName
        app.Status.DriverInfo.WebUIPort = service.nodePort
        // Create UI Ingress if ingress-format is set.
        if c.ingressURLFormat != "" {
            ingress, err := createSparkUIIngress(app, *service, c.ingressURLFormat, c.kubeClient)
            if err != nil {
                glog.Errorf("failed to create UI Ingress for SparkApplication %s/%s: %v", app.Namespace, app.Name, err)
            } else {
                app.Status.DriverInfo.WebUIIngressAddress = ingress.ingressURL
                app.Status.DriverInfo.WebUIIngressName = ingress.ingressName
            }
        }
    }
    return app
}
AI 代码解读

其实到此,我们就已经基本了解Spark Operator做的事情了,首先定义了两种不同的CRD对象,分别对应普通的计算任务与定时周期性的计算任务,然后解析CRD的配置文件,拼装成为spark-submit的命令,通过prometheus暴露监控数据采集接口,创建Service提供spark-ui的访问。然后通过监听Pod的状态,不断回写更新CRD对象,实现了spark作业任务的生命周期管理。

[](https://www.atatech.org/articles/136659#2)Spark Operator的任务状态机

当我们了解了Spark Operator的设计思路和基本流程后,还需要深入了解的就是sparkapplication的状态都包含哪些,他们之间是如何进行转换的,因为这是Spark Operator对于生命周期管理增强最重要的部分。

一个Spark的作业任务可以通过上述的状态机转换图进行表示,一个正常的作业任务经历如下几个状态:

New -> Submitted -> Running -> Succeeding -> Completed
AI 代码解读

而当任务失败的时候会进行重试,若重试超过最大重试次数则会失败。也就是说如果在任务的执行过程中,由于资源、调度等因素造成Pod被驱逐或者移除,Spark Operator都会通过自身的状态机状态转换进行重试。

[](https://www.atatech.org/articles/136659#3)Spark Operator的状态排查

我们已经知道了Spark Operator最核心的功能就是将CRD的配置转换为spark-submit的命令,那么当一个作业运行不预期的时候,我们该如何判断是哪一层出现的问题呢?首先我们要判断的就是spark-submit时所生成的参数是否是预期的,因为CRD的Yaml配置虽然可以增强表达能力,但是提高了配置的难度与出错的可能性。

func runSparkSubmit(submission *submission) (bool, error) {
    sparkHome, present := os.LookupEnv(sparkHomeEnvVar)
    if !present {
        glog.Error("SPARK_HOME is not specified")
    }
    var command = filepath.Join(sparkHome, "/bin/spark-submit")
    cmd := execCommand(command, submission.args...)
    glog.V(2).Infof("spark-submit arguments: %v", cmd.Args)
    if _, err := cmd.Output(); err != nil {
        var errorMsg string
        if exitErr, ok := err.(*exec.ExitError); ok {
            errorMsg = string(exitErr.Stderr)
        }
        // The driver pod of the application already exists.
        if strings.Contains(errorMsg, podAlreadyExistsErrorCode) {
            glog.Warningf("trying to resubmit an already submitted SparkApplication %s/%s", submission.namespace, submission.name)
            return false, nil
        }
        if errorMsg != "" {
            return false, fmt.Errorf("failed to run spark-submit for SparkApplication %s/%s: %s", submission.namespace, submission.name, errorMsg)
        }
        return false, fmt.Errorf("failed to run spark-submit for SparkApplication %s/%s: %v", submission.namespace, submission.name, err)
    }
    return true, nil
}
AI 代码解读

默认情况下Spark Operator会通过glog level=2等级对外输出每次作业提交后转换的提交命令。而默认情况下,glog的level即为2,因此通过检查Spark Operator的Pod日志可以协助开发者快速排查问题。此外在sparkapplication上面也会通过event的方式进行状态的记录,上述状态机之间的转换都会通过event的方式体现在sparkapplication的对象上。掌握这两种方式进行问题排查,可以节省大量排错时间。

[](https://www.atatech.org/articles/136659#4)最后

使用Spark Operator是在kubernetes上实践spark的最佳方式,和传统的spark-submit相比提供了更多的故障恢复与可靠性保障,并且提供了监控、日志、UI等能力的集成与支持。在下一篇中,会为大家介绍在kubernetes集群中,提交spark作业时的如何使用外部存储存储的最佳实践。

相关实践学习
通过Ingress进行灰度发布
本场景您将运行一个简单的应用,部署一个新的应用用于新的发布,并通过Ingress能力实现灰度发布。
容器应用与集群管理
欢迎来到《容器应用与集群管理》课程,本课程是“云原生容器Clouder认证“系列中的第二阶段。课程将向您介绍与容器集群相关的概念和技术,这些概念和技术可以帮助您了解阿里云容器服务ACK/ACK Serverless的使用。同时,本课程也会向您介绍可以采取的工具、方法和可操作步骤,以帮助您了解如何基于容器服务ACK Serverless构建和管理企业级应用。 学习完本课程后,您将能够: 掌握容器集群、容器编排的基本概念 掌握Kubernetes的基础概念及核心思想 掌握阿里云容器服务ACK/ACK Serverless概念及使用方法 基于容器服务ACK Serverless搭建和管理企业级网站应用
相关文章
HashMap深度解析:从原理到实战
HashMap,作为Java集合框架中的一个核心组件,以其高效的键值对存储和检索机制,在软件开发中扮演着举足轻重的角色。作为一名资深的AI工程师,深入理解HashMap的原理、历史、业务场景以及实战应用,对于提升数据处理和算法实现的效率至关重要。本文将通过手绘结构图、流程图,结合Java代码示例,全方位解析HashMap,帮助读者从理论到实践全面掌握这一关键技术。
96 14
深入解析云计算中的微服务架构:原理、优势与实践
深入解析云计算中的微服务架构:原理、优势与实践
106 3
自注意力机制全解析:从原理到计算细节,一文尽览!
自注意力机制(Self-Attention)最早可追溯至20世纪70年代的神经网络研究,但直到2017年Google Brain团队提出Transformer架构后才广泛应用于深度学习。它通过计算序列内部元素间的相关性,捕捉复杂依赖关系,并支持并行化训练,显著提升了处理长文本和序列数据的能力。相比传统的RNN、LSTM和GRU,自注意力机制在自然语言处理(NLP)、计算机视觉、语音识别及推荐系统等领域展现出卓越性能。其核心步骤包括生成查询(Q)、键(K)和值(V)向量,计算缩放点积注意力得分,应用Softmax归一化,以及加权求和生成输出。自注意力机制提高了模型的表达能力,带来了更精准的服务。
探索阿里云 Flink 物化表:原理、优势与应用场景全解析
阿里云Flink的物化表是流批一体化平台中的关键特性,支持低延迟实时更新、灵活查询性能、无缝流批处理和高容错性。它广泛应用于电商、物联网和金融等领域,助力企业高效处理实时数据,提升业务决策能力。实践案例表明,物化表显著提高了交易欺诈损失率的控制和信贷审批效率,推动企业在数字化转型中取得竞争优势。
97 16
探索网络模型与协议:从OSI到HTTPs的原理解析
OSI七层网络模型和TCP/IP四层模型是理解和设计计算机网络的框架。OSI模型包括物理层、数据链路层、网络层、传输层、会话层、表示层和应用层,而TCP/IP模型则简化为链路层、网络层、传输层和 HTTPS协议基于HTTP并通过TLS/SSL加密数据,确保安全传输。其连接过程涉及TCP三次握手、SSL证书验证、对称密钥交换等步骤,以保障通信的安全性和完整性。数字信封技术使用非对称加密和数字证书确保数据的机密性和身份认证。 浏览器通过Https访问网站的过程包括输入网址、DNS解析、建立TCP连接、发送HTTPS请求、接收响应、验证证书和解析网页内容等步骤,确保用户与服务器之间的安全通信。
117 3
Vue.js响应式原理深度解析:从Vue 2到Vue 3的演进
Vue.js响应式原理深度解析:从Vue 2到Vue 3的演进
105 17
深入解析区块链技术的核心原理与应用前景
深入解析区块链技术的核心原理与应用前景
83 0
Spark on Kubernetes 的现状与挑战
被称为云上 OS 的 Kubernetes 是 Cloud Native 理念的一种技术承载与体现,但是如何通过 Kubernetes 来助力大数据应用还是有很多可以探索的地方。欢迎交流。
Spark on Kubernetes 的现状与挑战
云原生时代,Kubernetes 的重要性日益凸显,这篇文章以 Spark 为例来看一下大数据生态 on Kubernetes 生态的现状与挑战。
2353 0
ClickHouse与大数据生态集成:Spark & Flink 实战
【10月更文挑战第26天】在当今这个数据爆炸的时代,能够高效地处理和分析海量数据成为了企业和组织提升竞争力的关键。作为一款高性能的列式数据库系统,ClickHouse 在大数据分析领域展现出了卓越的能力。然而,为了充分利用ClickHouse的优势,将其与现有的大数据处理框架(如Apache Spark和Apache Flink)进行集成变得尤为重要。本文将从我个人的角度出发,探讨如何通过这些技术的结合,实现对大规模数据的实时处理和分析。
214 2
ClickHouse与大数据生态集成:Spark & Flink 实战

云原生

+关注

相关产品

  • 容器服务Kubernetes版
  • 推荐镜像

    更多
    AI助理

    你好,我是AI助理

    可以解答问题、推荐解决方案等