Container Service for Kubernetes：Spark作業配置動態資源分派

根據如下內容建立SparkApplication資訊清單檔並儲存為spark-pagerank-celeborn-dra.yaml：

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: spark-pagerank-celeborn-dra
  namespace: spark
spec:
  type: Scala
  mode: cluster
  # 需將 <SPARK_IMAGE> 替換成您自己的Spark鏡像，該鏡像中需要包含JindoSDK依賴。
  image: <SPARK_IMAGE>
  mainApplicationFile: local:///opt/spark/examples/jars/spark-examples_2.12-3.5.4.jar
  mainClass: org.apache.spark.examples.SparkPageRank
  arguments:
  # 需將 <OSS_BUCKET> 替換成您的 OSS Bucket名稱。
  - oss://<OSS_BUCKET>/data/pagerank_dataset.txt
  - "10"
  sparkVersion: 3.5.4
  hadoopConf:
    # 支援使用 oss:// 格式訪問 OSS 資料。
    fs.AbstractFileSystem.oss.impl: org.apache.hadoop.fs.aliyun.oss.OSS
    fs.oss.impl: org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem
    # OSS 訪問端點，需將 <OSS_ENDPOINT> 替換成您的 OSS 訪問端點
    # 例如，北京地區 OSS 內網訪問端點為 oss-cn-beijing-internal.aliyuncs.com
    fs.oss.endpoint: <OSS_ENDPOINT>
    # 從環境變數中讀取 OSS 訪問憑據
    fs.oss.credentials.provider: com.aliyun.oss.common.auth.EnvironmentVariableCredentialsProvider
  sparkConf:
    # ====================
    # 事件記錄
    # ====================
    spark.eventLog.enabled: "true"
    spark.eventLog.dir: file:///mnt/nas/spark/event-logs

    # ====================
    # Celeborn
    # Ref: https://github.com/apache/celeborn/blob/main/README.md#spark-configuration
    # ====================
    # Shuffle manager class name changed in 0.3.0:
    #    before 0.3.0: `org.apache.spark.shuffle.celeborn.RssShuffleManager`
    #    since 0.3.0: `org.apache.spark.shuffle.celeborn.SparkShuffleManager`
    spark.shuffle.manager: org.apache.spark.shuffle.celeborn.SparkShuffleManager
    # Must use kryo serializer because java serializer do not support relocation
    spark.serializer: org.apache.spark.serializer.KryoSerializer
    # 需要根據 Celeborn master 副本數量進行配置。
    spark.celeborn.master.endpoints: celeborn-master-0.celeborn-master-svc.celeborn.svc.cluster.local,celeborn-master-1.celeborn-master-svc.celeborn.svc.cluster.local,celeborn-master-2.celeborn-master-svc.celeborn.svc.cluster.local
    # options: hash, sort
    # Hash shuffle writer use (partition count) * (celeborn.push.buffer.max.size) * (spark.executor.cores) memory.
    # Sort shuffle writer uses less memory than hash shuffle writer, if your shuffle partition count is large, try to use sort hash writer.
    spark.celeborn.client.spark.shuffle.writer: hash
    # We recommend setting `spark.celeborn.client.push.replicate.enabled` to true to enable server-side data replication
    # If you have only one worker, this setting must be false 
    # If your Celeborn is using HDFS, it's recommended to set this setting to false
    spark.celeborn.client.push.replicate.enabled: "false"
    # Support for Spark AQE only tested under Spark 3
    spark.sql.adaptive.localShuffleReader.enabled: "false"
    # we recommend enabling aqe support to gain better performance
    spark.sql.adaptive.enabled: "true"
    spark.sql.adaptive.skewJoin.enabled: "true"
    # 當 Spark 版本 >= 3.5.0 時，配置該選項以支援動態資源分派
    spark.shuffle.sort.io.plugin.class: org.apache.spark.shuffle.celeborn.CelebornShuffleDataIO
    spark.executor.userClassPathFirst: "false"

    # ====================
    # 動態資源分派
    # Ref: https://spark.apache.org/docs/latest/job-scheduling.html#dynamic-resource-allocation
    # ====================
    # 啟用動態資源分派
    spark.dynamicAllocation.enabled: "true"
    # 啟用 shuffle 檔案跟蹤，不依賴 ESS 即可實現動態資源分派。
    # 在使用 Celeborn 作為 RSS 時，當 Spark 版本 >= 3.4.0 時，強烈建議關閉該選項。
    spark.dynamicAllocation.shuffleTracking.enabled: "false"
    # Executor 數量的初始值。
    spark.dynamicAllocation.initialExecutors: "1"
    # Executor 數量的最小值。
    spark.dynamicAllocation.minExecutors: "0"
    # Executor 數量的最大值。
    spark.dynamicAllocation.maxExecutors: "5"
    # Executor 空閑逾時時間，超過該時間將會被釋放掉。
    spark.dynamicAllocation.executorIdleTimeout: 60s
    # 緩衝了資料區塊的 Executor 空閑逾時時間，超過該時間將會被釋放掉，預設為 infinity，即不會釋放。
    # spark.dynamicAllocation.cachedExecutorIdleTimeout:
    # 當存在待調度任務超過該時間後，將會申請更多的 executor。
    spark.dynamicAllocation.schedulerBacklogTimeout: 1s
    # 每間隔該時間後，將會開始下一批次申請 Executor。 
    spark.dynamicAllocation.sustainedSchedulerBacklogTimeout: 1s
  driver:
    cores: 1
    coreLimit: 1200m
    memory: 512m
    envFrom:
    - secretRef:
        name: spark-oss-secret
    volumeMounts:
    - name: nas
      mountPath: /mnt/nas
    serviceAccount: spark-operator-spark
  executor:
    cores: 1
    coreLimit: "1"
    memory: 4g
    envFrom:
    - secretRef:
        name: spark-oss-secret
    volumeMounts:
    - name: nas
      mountPath: /mnt/nas
  volumes:
  - name: nas
    persistentVolumeClaim:
      claimName: nas-pvc
  restartPolicy:
    type: Never

ARG SPARK_IMAGE=spark:3.5.4

FROM ${SPARK_IMAGE}

# Add dependency for Hadoop Aliyun OSS support
ADD --chown=spark:spark --chmod=644 https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aliyun/3.3.4/hadoop-aliyun-3.3.4.jar ${SPARK_HOME}/jars
ADD --chown=spark:spark --chmod=644 https://repo1.maven.org/maven2/com/aliyun/oss/aliyun-sdk-oss/3.17.4/aliyun-sdk-oss-3.17.4.jar ${SPARK_HOME}/jars
ADD --chown=spark:spark --chmod=644 https://repo1.maven.org/maven2/org/jdom/jdom2/2.0.6.1/jdom2-2.0.6.1.jar ${SPARK_HOME}/jars

# Add dependency for Celeborn
ADD --chown=spark:spark --chmod=644 https://repo1.maven.org/maven2/org/apache/celeborn/celeborn-client-spark-3-shaded_2.12/0.5.3/celeborn-client-spark-3-shaded_2.12-0.5.3.jar ${SPARK_HOME}/jars

執行如下命令提交SparkApplication。

kubectl apply -f spark-pagerank-celeborn-dra.yaml

預期輸出：

sparkapplication.sparkoperator.k8s.io/spark-pagerank-celeborn-dra created

查看Driver日誌。

kubectl logs -n spark spark-pagerank-celeborn-dra-driver | grep -a2 -b2 "Going to request"

預期輸出：

3544-25/01/16 03:51:28 INFO SparkKubernetesClientFactory: Auto-configuring K8S client using current context from users K8S config file
3674-25/01/16 03:51:30 INFO Utils: Using initial executors = 1, max of spark.dynamicAllocation.initialExecutors, spark.dynamicAllocation.minExecutors and spark.executor.instances
3848:25/01/16 03:51:30 INFO ExecutorPodsAllocator: Going to request 1 executors from Kubernetes for ResourceProfile Id: 0, target: 1, known: 0, sharedSlotFromPendingPods: 2147483647.
4026-25/01/16 03:51:30 INFO ExecutorPodsAllocator: Found 0 reusable PVCs from 0 PVCs
4106-25/01/16 03:51:30 INFO CelebornShuffleDataIO: Loading CelebornShuffleDataIO
--
11796-25/01/16 03:51:41 INFO TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0) (192.168.95.163, executor 1, partition 0, PROCESS_LOCAL, 9807 bytes)
11944-25/01/16 03:51:42 INFO BlockManagerInfo: Added broadcast_1_piece0 in memory on 192.168.95.163:37665 (size: 13.3 KiB, free: 2.1 GiB)
12076:25/01/16 03:51:42 INFO ExecutorPodsAllocator: Going to request 1 executors from Kubernetes for ResourceProfile Id: 0, target: 2, known: 1, sharedSlotFromPendingPods: 2147483647.
12254-25/01/16 03:51:42 INFO ExecutorAllocationManager: Requesting 1 new executor because tasks are backlogged (new desired total will be 2 for resource profile id: 0)
12416-25/01/16 03:51:42 INFO ExecutorPodsAllocator: Found 0 reusable PVCs from 0 PVCs

可以看到driver在03:51:30和03:51:42 時分別申請了1個Executor資源。

訪問Spark History Server查看，訪問步驟請參見訪問Spark History Server Web UI。

通過查看該作業的事件時間軸，我們也可以看到兩個executor先後被建立。