Container Service for Kubernetes:Spark ジョブの動的リソース割り当てを構成する

spark-pagerank-celeborn-dra.yaml という名前のファイルを作成し、次の内容をファイルにコピーします。このファイルは、SparkApplication を作成するために使用されます。

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 バケットの名前に置き換えます。
  - 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
    # 参照: https://github.com/apache/celeborn/blob/main/README.md#spark-configuration
    # ====================
    # シャッフルマネージャのクラス名は 0.3.0 で変更されました。
    #    0.3.0 より前: `org.apache.spark.shuffle.celeborn.RssShuffleManager`
    #    0.3.0 以降: `org.apache.spark.shuffle.celeborn.SparkShuffleManager`
    spark.shuffle.manager: org.apache.spark.shuffle.celeborn.SparkShuffleManager
    # Java シリアライザは再配置をサポートしていないため、kryo シリアライザを使用する必要があります
    spark.serializer: org.apache.spark.serializer.KryoSerializer
    # Celeborn マスターノードのレプリカ数に基づいて、このパラメータを構成します。
    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
    # オプション: hash, sort
    # ハッシュシャッフルライターは、(パーティション数) * (celeborn.push.buffer.max.size) * (spark.executor.cores) のメモリを使用します。
    # シャッフルパーティション数が大きい場合、ソートハッシュライターはハッシュシャッフルライターよりも少ないメモリを使用するため、ソートハッシュライターを使用してみてください。
    spark.celeborn.client.spark.shuffle.writer: hash
    # サーバー側のデータレプリケーションを有効にするには、`spark.celeborn.client.push.replicate.enabled` を true に設定することをお勧めします
    # ワーカーが 1 つしかない場合は、この設定を false にする必要があります
    # Celeborn が HDFS を使用している場合は、この設定を false に設定することをお勧めします
    spark.celeborn.client.push.replicate.enabled: "false"
    # Spark AQE のサポートは、Spark 3 でのみテストされています
    spark.sql.adaptive.localShuffleReader.enabled: "false"
    # パフォーマンスを向上させるために、aqe サポートを有効にすることをお勧めします
    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"

    # ====================
    # 動的リソース割り当て
    # 参照: https://spark.apache.org/docs/latest/job-scheduling.html#dynamic-resource-allocation
    # ====================
    # 動的リソース割り当てを有効にする
    spark.dynamicAllocation.enabled: "true"
    # シャッフルファイルの追跡を有効にし、ESS に依存せずに動的リソース割り当てを実装します。
    # Spark のバージョンが 3.4.0 以降の場合、Celeborn を RSS として使用するときにこの機能を無効にすることをお勧めします。
    spark.dynamicAllocation.shuffleTracking.enabled: "false"
    # エグゼキュータ数の初期値。
    spark.dynamicAllocation.initialExecutors: "1"
    # エグゼキュータの最小数。
    spark.dynamicAllocation.minExecutors: "0"
    # エグゼキュータの最大数。
    spark.dynamicAllocation.maxExecutors: "5"
    # エグゼキュータのアイドルタイムアウト期間。タイムアウト期間を超えると、エグゼキュータは解放されます。
    spark.dynamicAllocation.executorIdleTimeout: 60s
    # データブロックをキャッシュしたエグゼキュータのアイドルタイムアウト期間。タイムアウト期間を超えると、エグゼキュータは解放されます。デフォルト値は infinity で、データブロックが解放されないことを指定します。
    # spark.dynamicAllocation.cachedExecutorIdleTimeout:
    # ジョブが指定されたスケジュール時間を超えると、追加のエグゼキュータが要求されます。
    spark.dynamicAllocation.schedulerBacklogTimeout: 1s
    # 各時間間隔の後、後続のバッチのエグゼキュータが要求されます。
    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}

# Hadoop Aliyun OSS サポートの依存関係を追加します
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

# 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

ドライバログを表示します。

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

コマンド出力は、ドライバが 03:51:30 に 1 つのエグゼキュータリソースを、03:51:42 に別のエグゼキュータリソースを要求していることを示しています。

Spark History Server を使用してログを表示します。詳細については、「Spark History Server を使用して Spark ジョブに関する情報を表示する」をご参照ください。

ジョブのイベントタイムラインは、2 つのエグゼキュータが順番に作成されていることを示しています。