Spark作业的多集群调度与分发

Apache Spark是一种专门用于大规模数据处理的计算引擎，广泛应用于数据分析和机器学习等场景。本文介绍如何通过ACK One舰队将Spark作业在多集群中进行调度和分发运行，帮助您提升多集群中空闲资源利用率。

工作原理

为关联集群安装ack-spark-operator组件。
为舰队创建SparkApplication和PropagationPolicy。
舰队的多集群调度组件（Global Scheduler）会根据每个关联子集群的剩余资源，来匹配Spark作业资源请求完成调度。
对集群版本为1.28及以上的子集群，舰队支持通过资源预占来提高Spark作业调度的成功率。
舰队调度完作业后，通过多集群应用分发SparkApplication至调度选中的关联集群中。
在关联集群中，ACK Spark Operator负责运行Spark作业的driver和executor。同时舰队会监视子集群中Spark作业运行状态，若driver由于资源不足而无法运行，舰队会在一定时间后回收该SparkApplication，并重调度到其他资源充足的关联集群中。

前提条件

舰队实例已添加多个关联集群，且关联集群版本为1.18及以上。具体操作，请参见管理关联集群。
已授予RAM用户AliyunAdcpFullAccess权限。具体操作，请参见为RAM用户授权。
已安装AMC命令行工具。

步骤一：为关联集群安装ack-spark-operator组件

请先在期望运行Spark作业的子集群上安装ack-spark-operator组件。

登录容器服务管理控制台，在左侧导航栏选择市场 > 应用市场。
在应用市场页面单击应用目录页签，然后搜索并选中ack-spark-operator。
在ack-spark-operator页面，单击一键部署。
在创建面板中，选择集群和命名空间，然后单击下一步。

在参数配置页面，设置相应参数，然后单击确定。

下表列出了部分配置参数的说明。完整的参数配置详情，您可以在ack-spark-operator页面中的配置项查看。

参数	描述	示例值
`controller.replicas`	控制器副本数量。	1（默认值）
`webhook.replicas`	Webhook副本数量。	1（默认值）
`spark.jobNamespaces`	可运行Spark任务的命名空间列表。包含空字符串表示允许所有命名空间。多个命名空间使用英文半角逗号 `,` 隔开。	`["default"]`（默认值） `[""]`（所有命名空间） `["ns1","ns2","ns3"]`（多个命名空间）
`spark.serviceAccount.name`	Spark作业会在`spark.jobNamespaces`指定的每个命名空间中自动创建名为`spark-operator-spark`的ServiceAccount和RBAC资源并进行相关授权。您可以自定义ServiceAccount名称，后续提交Spark作业时请指定自定义创建的ServiceAccount名称。	`spark-operator-spark`（默认值）

步骤二：在舰队上创建PriorityClass，并分发到子集群

为了保证提交的Spark作业不会抢占在线服务资源、影响在线服务产品的正常运行，建议对要提交的Spark作业设置低优先级。

使用舰队的KubeConfig，创建一个低优先级的PriorityClass，其value设置为负数。

apiVersion: scheduling.k8s.io/v1
kind: PriorityClass
metadata:
  name: low-priority
value: -1000
globalDefault: false
description: "Low priority for Spark applications"

在舰队中创建ClusterPropagationPolicy，将PriorityClass分发到指定集群中。若您希望将PriorityClass分发到所有关联集群，可以将clusterAffinity部分删除。

apiVersion: policy.one.alibabacloud.com/v1alpha1
kind: ClusterPropagationPolicy
metadata:
  name: priority-policy
spec:
  preserveResourcesOnDeletion: false
  resourceSelectors:
  - apiVersion: scheduling.k8s.io/v1
    kind: PriorityClass
  placement:
    clusterAffinity:
      clusterNames:
      - ${cluster1-id} # 您的集群ID。
      - ${cluster2-id} # 您的集群ID。
#      labelSelector:
#        matchLabels:
#          key: value
    replicaScheduling:
      replicaSchedulingType: Duplicated

步骤三：在舰队上提交Spark作业，并调度到子集群

（可选）在舰队中创建Namespace，并下发到子集群。
1. 如果待分发应用所在的命名空间在舰队中不存在，则需要先在舰队实例中创建该命名空间，并确保步骤一中安装组件的spark.jobNamespaces中包含该命名空间。如果命名空间已存在，则可以跳过此步骤。
  使用舰队的KubeConfig，执行以下命令，创建相应命名空间。
```
kubectl create ns xxx
```
2. 子集群也需要相应的命名空间。如果尚未创建，可以使用ClusterPropagationPolicy将舰队中的命名空间下发至各个子集群。
```
apiVersion: policy.one.alibabacloud.com/v1alpha1
kind: ClusterPropagationPolicy
metadata:
  name: ns-policy
spec:
  resourceSelectors:
  - apiVersion: v1
    kind: Namespace
    name: xxx
  placement:
    clusterAffinity:
      clusterNames:
      - ${cluster1-id} # 您的集群ID。
      - ${cluster2-id} # 您的集群ID。
    replicaScheduling:
      replicaSchedulingType: Duplicated
```

在舰队中创建如下的PropagationPolicy，即可将所有sparkoperator.k8s.io/v1beta2的SparkApplication资源下发至相应的集群中。

apiVersion: policy.one.alibabacloud.com/v1alpha1
kind: PropagationPolicy
metadata:
  name: sparkapp-policy 
  namespace: default
spec:
  preserveResourcesOnDeletion: false
  propagateDeps: true
  placement:
    clusterAffinity:
      clusterNames:
      - ${cluster1-id} # 您的集群ID。
      - ${cluster2-id} # 您的集群ID。
#      labelSelector:
#        matchLabels:
#          key: value
    replicaScheduling:
      replicaSchedulingType: Divided
      customSchedulingType: Gang
  resourceSelectors:
    - apiVersion: sparkoperator.k8s.io/v1beta2
      kind: SparkApplication

在舰队中创建SparkApplication，并在driver和executor中指定priorityClassName。创建完成后，该应用将由第2步中的PropagationPolicy分发到调度选中的集群中。

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: spark-pi
  namespace: default     # 需要确保命名空间在spark.jobNamespaces指定的命名空间列表中。
spec:
  type: Scala
  mode: cluster
  image: registry-cn-hangzhou.ack.aliyuncs.com/ack-demo/spark:3.5.4
  imagePullPolicy: IfNotPresent
  mainClass: org.apache.spark.examples.SparkPi
  mainApplicationFile: local:///opt/spark/examples/jars/spark-examples_2.12-3.5.4.jar
  arguments:
  - "1000"
  sparkVersion: 3.5.4
  driver:
    cores: 1
    memory: 512m
    priorityClassName: low-priority
    serviceAccount: spark-operator-spark   # 如果您自定义了ServiceAccount名称，则需要进行相应修改。
  executor:
    instances: 1
    cores: 1
    memory: 512m
    priorityClassName: low-priority
  restartPolicy:
    type: Never

步骤四：查看Spark作业

在舰队中执行以下命令，查看舰队中Spark作业的状态。

kubectl get sparkapp

预期输出：

NAME       STATUS    ATTEMPTS   START                  FINISH       AGE
spark-pi   RUNNING   1          2025-02-24T12:10:34Z   <no value>   11s

在舰队中执行以下命令，在事件中查看Spark作业调度至哪个关联集群。

kubectl describe sparkapp spark-pi

预期输出：

 Normal   ScheduleBindingSucceed  2m29s                  default-scheduler                   Binding has been scheduled successfully. Result: {c6xxxxx:0,[{driver 1} {executor 1}]}

在舰队中执行以下命令，查看关联集群中Spark作业的状态。

kubectl amc get sparkapp -M

预期输出：

NAME       CLUSTER     STATUS      ATTEMPTS   START                  FINISH                 AGE   ADOPTION
spark-pi   c6xxxxxxx   COMPLETED   1          2025-02-24T12:10:34Z   2025-02-24T12:11:20Z   61s   Y

在舰队中执行以下命令，查看Pod状态。

kubectl amc get pod -M

预期输出：

NAME              CLUSTER     READY   STATUS      RESTARTS   AGE
spark-pi-driver   c6xxxxxxx   0/1     Completed   0          68s

在舰队中执行以下命令，查看关联集群中Spark作业的详情。
```
kubectl amc get sparkapp spark-pi -m ${member clusterid} -oyaml
```

工作原理

前提条件

步骤一：为关联集群安装ack-spark-operator组件

步骤二：在舰队上创建PriorityClass，并分发到子集群

步骤三：在舰队上提交Spark作业，并调度到子集群

步骤四：查看Spark作业

如何在舰队集群中使用KubeQueue并通过Gang调度PyTorchJob 2025-04-18 18:10

Spark作业的多集群调度与分发 2025-04-18 18:10

基于实际剩余资源的多集群Spark作业调度与分发 2025-04-18 18:10

目录