如何Kubernetes + ECI 部署 Spark 作业体验极致弹性?

背景 Kubernetes的优势 Spark on kubernetes相比于on YARN等传统部署方式的优势：

1、统一的资源管理。不论是什么类型的作业都可以在一个统一kubernetes的集群运行。不再需要单独为大数据作业维护一个独立的YARN集群。

2、弹性的集群基础设施。资源层和应用层提供了丰富的弹性策略，我们可以根据应用负载需求选择 ECS 虚拟机、神龙裸金属和 GPU 实例进行扩容，除了kubernetes集群本生具备的强大的扩缩容能力，还可以对接生态，比如virtual kubelet。

3、轻松实现复杂的分布式应用的资源隔离和限制，从YRAN复杂的队列管理和队列分配中解脱。

4、容器化的优势。每个应用都可以通过docker镜像打包自己的依赖，运行在独立的环境，甚至包括Spark的版本，所有的应用之间都是隔离的。

5、大数据上云。目前大数据应用上云常见的方式有两种：1）用ECS自建YARN（不限于YARN）集群；2）购买EMR服务。如今多了一个选择——Kubernetes。既能获得完全的集群级别的掌控，又能从复杂的集群管理、运维中解脱，还能享受云所带来的弹性和成本优势。

Spark自2.3.0开始试验性支持Standalone、on YARN以及on Mesos之外的新的部署方式：Running Spark on Kubernetes ，并在后续的发行版中不断地加强。

后文将是实际的操作，分别让Spark应用在普通的Kubernetes集群、Serverless Kubernetes集群、以及Kubernetes + virtual kubelet等三种场景中部署并运行。

Spark on Kubernetes 准备数据以及Spark应用镜像 参考： 在ECI中访问HDFS的数据

在ECI中访问OSS的数据

创建kubernetes集群 如果已经有阿里云的ACK集群，该步可以忽略。

具体的创建流程参考：创建Kubernetes 托管版集群。

提交作业 为Spark创建一个RBAC的role 创建账号(默认namespace)

kubectl create serviceaccount spark 绑定角色

kubectl create clusterrolebinding spark-role --clusterrole=edit --serviceaccount=default:spark --namespace=default 直接使用spark-submit提交（不推荐的提交方式） liumihustdeMacBook-Pro:spark-on-k8s liumihust$ ./spark-2.3.0-bin-hadoop2.6/bin/spark-submit --master k8s://121.199.47.XX:6443 --deploy-mode cluster --name WordCount --class com.aliyun.liumi.spark.example.WordCount --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark --conf spark.executor.instances=2 --conf spark.kubernetes.container.image=registry.cn-beijing.aliyuncs.com/liumi/spark:2.4.4-example local:///opt/spark/jars/SparkExampleJava-1.0-SNAPSHOT.jar 参数解释 —master ：k8s集群的apiserver，这是决定spark是在k8s集群跑，还是在yarn上跑。

—deploy-mode：driver可以部署在集群的master节点（client）也可以在非master（cluster）节点。

spark.executor.instances: executor的数量

spark.kubernetes.container.image spark打包镜像（包含driver、excutor、应用，也支持单独配置）

提交基本流程 spark-10.png

Running Spark on Kubernetes Spark先在k8s集群中创建Spark Driver（pod）。

Driver起来后，调用k8s API创建Executors（pods），Executors才是执行作业的载体。

作业计算结束，Executor Pods会被自动回收，Driver Pod处于Completed状态（终态）。可以供用户查看日志等。

Driver Pod只能被用户手动清理，或者被k8s GC回收。

结果分析 执行过程中的截图如下：spark-5.png

我们30G的数据用2个1C1G的Excutor处理了大约20分钟。

作业运行结束后查看结果：

[root@liumi-hdfs ~]# $HADOOP_HOME/bin/hadoop fs -cat /pod/data/A-Game-of-Thrones-Result/* (142400000,the) (78400000,and) (77120000,) (62200000,to) (56690000,of) (56120000,a) (43540000,his) (35160000,was) (30480000,he) (29060000,in) (26640000,had) (26200000,her) (23050000,as) (22210000,with) (20450000,The) (19260000,you) (18300000,I) (17510000,she) (16960000,that) (16450000,He) (16090000,not) (15980000,it) (15080000,at) (14710000,for) (14410000,on) (12660000,but) (12470000,him) (12070000,is) (11240000,from) (10300000,my) (10280000,have) (10010000,were) 至此，已经能在kubernetes集群部署并运行spark作业。

Spark on Serverless Kubernetes Serverless Kubernetes (ASK) 相比于普通的kubernetes集群，比较大的一个优势是，提交作业前无需提前预留任何资源，无需关心集群的扩缩容，所有资源都是随作业提交自动开始申请，作业执行结束后自动释放。作业执行完后就只剩一个SparkApplication和终态的Driver pod（只保留管控数据）。原理图如下图所示：spark-7.png

Running Spark on Serverless Kubernetes ASK通过virtual kubelet调度pod到阿里云弹性容器实例。虽然架构上跟ACK有明显的差异，但是两者都是全面兼容kubernetes标准的。所以on ASK跟前面的spark on kubernetes准备阶段的基本是一致的，即HDFS数据准备，spark base镜像的准备、spark应用镜像的准备等。主要就是作业提交方式稍有不同，以及一些额外的基本环境配置。

创建serverless kubernetes集群 创建以及操作集群的详细步骤参考：操作Serverless Kubernetes集群的方式

本文都是拷贝kubeconfig到本地服务器来访问集群。

选择标准serverless集群：eci-spark-4

基本参数：

1、自定义集群名。

2、选择地域、以及可用区。

3、专有网络可以用已有的也可以由容器服务自动创建的。

4、是否公网暴露API server，如有需求建议开启。

5、开启privatezone，必须开启。

6、日志收集，建议开启。eci-spark-5

注： 1、提交之前一定要升级集群的集群的virtual kubelet的版本（新建的集群可以忽略），只有目前最新版的VK才能跑Spark作业。

2、ASK集群依赖privatezone做服务发现，所以集群不需要开启privatezone，创建的时候需要勾选。如果创建的时候没有勾选，需要联系我们帮开启。不然Spark excutor会找不到driver service。

*制作镜像cache 由于后面可能要进行大规模启动，为了提高容器启动速度，提前将Spark应用的镜像缓存到ECI本地，采用k8s标准的CRD的方式，具体的流程参考：使用CRD加速创建Pod

提交： 由于spark submit目前支持的参数非常有限，所以ASK场景中建议不要使用spark submit直接提交，而是直接采用Spark Operator。也是我们推荐的方式。

Spark Operator 就是为了解决在Kubernetes集群部署并维护Spark应用而开发的。

eci-spark-6

Spark Operator几个主要的概念：

SparkApplication：标准的k8s CRD，有CRD就有一个Controller 与之对应。Controller负责监听CRD的创建、更新、以及删除等事件，并作出对应的Action。

ScheduledSparkApplication：SparkApplication的升级，支持带有自定义时间调度策略的作业提交，比如cron。

Submission runner：对Controller发起的创建请求提交spark-submit。

Spark pod monitor：监听Spark pods的状态和事件更新并告知Controller。

安装Spark Operator 推荐用 helm 3.0

helm repo add incubator http://storage.googleapis.com/kubernetes-charts-incubator helm install incubator/sparkoperator --namespace default --set operatorImageName=registry.cn-hangzhou.aliyuncs.com/eci_open/spark-operator --set operatorVersion=v1beta2-1.0.1-2.4.4 --generate-name --set enableWebhook=true 注：在Serverless Kubernetes安装时不要使用enableWebhook=true选项 安装完成后可以看到集群多了个spark operator pod。eci-saprk-7

选项说明：

1、—set operatorImageName:指定operator镜像，默认的google的镜像阿里云ECI内拉不下来，可以先拉取到本地然后推到ACR。

2、—set operatorVersion operator:镜像仓库名和版本不要写在一起。

3、—generate-name 可以不用显式设置安装名。

4、—set enableWebhook 默认不会打开，对于需要使用ACK+ECI的用户，会用到nodeSelector、tolerations这些高级特性，Webhook 必须要打开，后面会讲到。Serverless Kubernetes 不要打开。

注： 创建spark operator的时候，一定要确保镜像能拉下来，推荐直接使用eci_open提供的镜像，因为spark operator卸载的时候也是用相同的镜像启动job进行清理，如果镜像拉不下来清理job也会卡主，导致所有的资源都要手动清理，比较麻烦。

申明wordcount SparkApplication:

apiVersion: "sparkoperator.k8s.io/v1beta2" kind: SparkApplication metadata: name: wordcount namespace: default spec: type: Java mode: cluster image: "registry.cn-beijing.aliyuncs.com/liumi/spark:2.4.4-example" imagePullPolicy: IfNotPresent mainClass: com.aliyun.liumi.spark.example.WordCount mainApplicationFile: "local:///opt/spark/jars/SparkExampleJava-1.0-SNAPSHOT.jar" sparkVersion: "2.4.4" restartPolicy: type: OnFailure onFailureRetries: 2 onFailureRetryInterval: 5 onSubmissionFailureRetries: 2 onSubmissionFailureRetryInterval: 10 timeToLiveSeconds: 36000 sparkConf: "spark.kubernetes.allocation.batch.size": "10" driver: cores: 2 memory: "4096m" labels: version: 2.4.4 spark-app: spark-wordcount role: driver annotations: k8s.aliyun.com/eci-image-cache: "true" serviceAccount: spark executor: cores: 1 instances: 100 memory: "1024m" labels: version: 2.4.4 role: executor annotations: k8s.aliyun.com/eci-image-cache: "true" 注：大部分的参数都可以直接通过SparkApplication CRD已经支持的参数设置，目前支持的所有参数参考:SparkApplication CRD，此外还支持直接以sparkConf形式的传入。

提交： kubectl create -f wordcount-operator-example.yaml 结果分析 我们是100个1C1G的Excutor并发启动，应用的镜像大小约为 500 MB。

作业执行过程截图：eci-spark-8eci-spark-9

可以看到并发启动的100个pod基本在30s内可以完成全部的启动，其中93%可以在20秒内完成启动。

看下作业执行时间（包括了vk调度100个Excutor pod时间、每个Excutor pod资源准备的时间、以及作业实际执行的时间等）:

exitCode: 0 finishedAt: '2019-11-16T07:31:59Z' reason: Completed startedAt: '2019-11-16T07:29:01Z' 可以看到总共只花了178S，时间降了一个数量级。

ACK + ECI 在Spark中，Driver和Excutor之间的启动顺序是串行的。尽管ECI展现了出色的并发创建Executor pod的能力，但是ASK这种特殊架构会让Driver和Excutor之间的这种串行体现的比较明显，通常情况下在ECI启动一个Driver pod需要大约20s的时间，然后才是大规模的Excutor pod的启动。对于一些响应要求高的应用，Driver的启动速度可能比Excutor执行作业的耗时更重要。这个时候，我们可以采用ACK+ECI，即传统的Kubernetes集群 + virtual kubelet的方式：eci-spark-9

对于用户来说，只需如下简单的几步就可以将excutor调度到ECI的virtual node。

1、在ACK集群中安装ECI的virtual kubelet。 进入容器服务控制台的应用目录栏，搜索”ack-virtual-node”：

eci-spark-10

点击进入，选择要安装的集群。eci-spark-11

必填参数参考：

virtualNode: image: repository: registry.cn-hangzhou.aliyuncs.com/acs/virtual-nodes-eci tag: v1.0.0.1-aliyun affinityAdminssion: enabled: true image: repository: registry.cn-hangzhou.aliyuncs.com/ask/virtual-node-affinity-admission-controller tag: latest env: ECI_REGION: "cn-hangzhou" #集群所在的地域 ECI_VPC: vpc-bp187fy2e7l123456 # 集群所在的vpc，和创建集群的时候保持一致即可，可以在集群概览页查看 ECI_VSWITCH: vsw-bp1bqf53ba123456 # 资源所在的交换机，同上 ECI_SECURITY_GROUP: sg-bp12ujq5zp12346 # 资源所在的安全组，同上 ECI_ACCESS_KEY: XXXXX #账号AK ECI_SECRET_KEY: XXXXX #账号SK ALIYUN_CLUSTERID: virtual-kubelet 2、修改应用的yaml 为excutor增加如下参数即可：

nodeSelector: type: virtual-kubelet tolerations: - key: virtual-kubelet.io/provider operator: Exists 完整的应用参数如下：

apiVersion: "sparkoperator.k8s.io/v1beta2" kind: SparkApplication metadata: name: wordcount namespace: default spec: type: Java mode: cluster image: "registry.cn-beijing.aliyuncs.com/liumi/spark:2.4.4-example" imagePullPolicy: IfNotPresent mainClass: com.aliyun.liumi.spark.example.WordCount mainApplicationFile: "local:///opt/spark/jars/SparkExampleJava-1.0-SNAPSHOT.jar" sparkVersion: "2.4.4" restartPolicy: type: OnFailure onFailureRetries: 2 onFailureRetryInterval: 5 onSubmissionFailureRetries: 2 onSubmissionFailureRetryInterval: 10 timeToLiveSeconds: 36000 sparkConf: "spark.kubernetes.allocation.batch.size": "10" driver: cores: 2 memory: "4096m" labels: version: 2.4.4 spark-app: spark-wordcount role: driver annotations: k8s.aliyun.com/eci-image-cache: "true" serviceAccount: spark executor: cores: 1 instances: 100 memory: "1024m" labels: version: 2.4.4 role: executor annotations: k8s.aliyun.com/eci-image-cache: "true" #nodeName: virtual-kubelet nodeSelector: type: virtual-kubelet tolerations: - key: virtual-kubelet.io/provider operator: Exists 这样就可以将Driver调度到ACK，Excutor调度到ECI上，完美互补。

3、提交 效果如下：eci-spark-12

看下作业执行时间：

exitCode: 0 finishedAt: '2019-11-16T07:25:05Z' reason: Completed startedAt: '2019-11-16T07:22:40Z' 总共花了145秒，更重要的是Driver直接在本地起，只花了约2秒的时间就启动了。

附录 Spark Base 镜像： 本样例采用的是谷歌提供的 gcr.io/spark-operator/spark:v2.4.4

ECI已经帮拉取到ACR仓库，各地域地址如下：

公网地址：registry.{对应regionId}.aliyuncs.com/eci_open/spark:2.4.4

vpc网络地址：registry-vpc.{对应regionId}.aliyuncs.com/eci_open/spark:2.4.4

Spark Operator 镜像 本样例采用的是谷歌提供的 gcr.io/spark-operator/spark-operator:v1beta2-1.0.1-2.4.4

ECI已经帮拉取到ACR仓库，各地域地址如下：

公网地址：registry.{对应regionId}.aliyuncs.com/eci_open/spark-operator:v1beta2-1.0.1-2.4.4

vpc网络地址：registry-vpc.{对应regionId}.aliyuncs.com/eci_open/spark-operator:v1beta2-1.0.1-2.4.4