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

工作原理

1. 为关联集群安装ack-spark-operator组件。

2. 为舰队创建SparkApplication和PropagationPolicy。

3. 舰队的多集群调度组件（Global Scheduler）会根据每个关联子集群的剩余资源，来匹配Spark作业资源请求完成调度。

   对集群版本为1.28及以上的子集群，舰队支持通过资源预占来提高Spark作业调度的成功率。

4. 舰队调度完作业后，通过多集群应用分发SparkApplication至调度选中的关联集群中。

5. 在关联集群中，ACK Spark Operator负责运行Spark作业的driver和executor。同时舰队会监视子集群中Spark作业运行状态，若driver由于资源不足而无法运行，舰队会在一定时间后回收该SparkApplication，并重调度到其他资源充足的关联集群中。

前提条件

• 舰队实例已添加多个关联集群，且关联集群版本为1.18及以上。具体操作，请参见管理关联集群。

• 已授予RAM用户AliyunAdcpFullAccess权限。具体操作，请参见为RAM用户授权。

• 已安装AMC命令行工具。具体操作，请参见AMC命令行帮助。

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

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

1. 登录容器服务管理控制台，在左侧导航栏选择市场 > 应用市场。

2. 在应用市场页面单击应用目录页签，然后搜索并选中ack-spark-operator。

3. 在ack-spark-operator页面，单击一键部署。

4. 在创建面板中，选择集群和命名空间，然后单击下一步。

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

   下表列出了部分配置参数的说明。完整的参数配置详情，您可以在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作业设置低优先级。

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

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

2. 在舰队中创建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作业，并调度到子集群

1. （可选）在舰队中创建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

2. 在舰队中创建如下的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

3. 在舰队中创建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作业

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

   kubectl get sparkapp

   预期输出：

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

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

   kubectl describe sparkapp spark-pi

   预期输出：

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

3. 在舰队中执行以下命令，查看关联集群中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

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

   kubectl amc get pod -M     

   预期输出：

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

5. 在舰队中执行以下命令，查看关联集群中Spark作业的详情。

   kubectl amc get sparkapp spark-pi -m ${member clusterid} -oyaml