Container Service for Kubernetes:ACK 上の Spark の概要

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

# log4j-layout-template-json の依存関係を追加します
ADD --chown=spark:spark --chmod=644 https://repo1.maven.org/maven2/org/apache/logging/log4j/log4j-layout-template-json/2.24.1/log4j-layout-template-json-2.24.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

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: spark-pi
  namespace: spark # この namespace が spark.jobNamespaces で指定された namespace リストに含まれていることを確認します。
spec:
  type: Scala
  mode: cluster
  # <SPARK_IMAGE> を独自の Spark コンテナイメージに置き換えます。
  image: <SPARK_IMAGE>
  imagePullPolicy: IfNotPresent
  mainClass: org.apache.spark.examples.SparkPi
  mainApplicationFile: local:///opt/spark/examples/jars/spark-examples_2.12-3.5.4.jar
  arguments:
  - "5000"
  sparkVersion: 3.5.4
  driver:
    cores: 1
    coreLimit: 1200m
    memory: 512m
    template:
      spec:
        containers:
        - name: spark-kubernetes-driver
        serviceAccount: spark-operator-spark
  executor:
    instances: 1
    cores: 1
    coreLimit: 1200m
    memory: 512m
    template:
      spec:
        containers:
        - name: spark-kubernetes-executor
  restartPolicy:
    type: Never

「Spark ジョブで OSS データを読み書きする」を参照して、テストデータセットを OSS にアップロードします。その後、次の Spark ジョブの例を実行できます。

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: spark-pagerank
  namespace: spark
spec:
  type: Scala
  mode: cluster
  # <SPARK_IMAGE> を独自の Spark イメージに置き換えます
  image: <SPARK_IMAGE>
  imagePullPolicy: IfNotPresent
  mainClass: org.apache.spark.examples.SparkPageRank
  mainApplicationFile: local:///opt/spark/examples/jars/spark-examples_2.12-3.5.4.jar
  arguments:
  # 入力テストデータセットを指定します。<OSS_BUCKET> を OSS バケット名に置き換えます
  - oss://<OSS_BUCKET>/data/pagerank_dataset.txt
  # 反復回数
  - "10"
  sparkVersion: 3.5.4
  hadoopConf:
    # OSS への Spark ジョブアクセスを構成します
    fs.oss.impl: org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem
    # <OSS_ENDPOINT> を OSS アクセスエンドポイントに置き換えます。たとえば、北京の OSS の内部ネットワークアクセスエンドポイントは oss-cn-beijing-internal.aliyuncs.com です
    fs.oss.endpoint: <OSS_ENDPOINT>
    fs.oss.credentials.provider: com.aliyun.oss.common.auth.EnvironmentVariableCredentialsProvider
  driver:
    cores: 1
    coreLimit: "1"
    memory: 4g
    template:
      spec:
        containers:
        - name: spark-kubernetes-driver
          envFrom:
          # 指定された Secret から環境変数を読み取ります
          - secretRef:
              name: spark-oss-secret
        serviceAccount: spark-operator-spark
  executor:
    instances: 2
    cores: 1
    coreLimit: "1"
    memory: 4g
    template:
      spec:
        containers:
        - name: spark-kubernetes-executor
          envFrom:
          # 指定された Secret から環境変数を読み取ります
          - secretRef:
              name: spark-oss-secret
  restartPolicy:
    type: Never

# Spark 構成
sparkConf:
  spark.history.fs.logDirectory: file:///mnt/nas/spark/event-logs

# 環境変数
env:
- name: SPARK_DAEMON_MEMORY
  value: 7g

# データボリューム
volumes:
- name: nas
  persistentVolumeClaim:
    claimName: nas-pvc

# データボリュームマウント
volumeMounts:
- name: nas
  subPath: spark/event-logs
  mountPath: /mnt/nas/spark/event-logs

# Spark ジョブの数と規模に基づいてリソースサイズを調整します
resources:
  requests:
    cpu: 2
    memory: 8Gi
  limits:
    cpu: 2
    memory: 8Gi

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: spark-pi
  namespace: spark
spec:
  type: Scala
  mode: cluster
  # <SPARK_IMAGE> を Spark イメージに置き換えます
  image: <SPARK_IMAGE>
  imagePullPolicy: IfNotPresent
  mainClass: org.apache.spark.examples.SparkPi
  mainApplicationFile: local:///opt/spark/examples/jars/spark-examples_2.12-3.5.4.jar
  arguments:
  - "5000"
  sparkVersion: 3.5.4
  sparkConf:
    # イベントログ
    spark.eventLog.enabled: "true"
    spark.eventLog.dir: file:///mnt/nas/spark/event-logs
  driver:
    cores: 1
    coreLimit: 1200m
    memory: 512m
    template:
      spec:
        containers:
        - name: spark-kubernetes-driver
          volumeMounts:
          - name: nas
            subPath: spark/event-logs
            mountPath: /mnt/nas/spark/event-logs
        volumes:
        - name: nas
          persistentVolumeClaim:
            claimName: nas-pvc
        serviceAccount: spark-operator-spark
  executor:
    instances: 1
    cores: 1
    coreLimit: 1200m
    memory: 512m
    template:
      spec:
        containers:
        - name: spark-kubernetes-executor
  restartPolicy:
    type: Never

このコードは、Spark ジョブで Simple Log Service を使用して、Spark コンテナの /opt/spark/logs/*.log にあるログを収集します。

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: spark-pi
  namespace: spark
spec:
  type: Scala
  mode: cluster
  # <SPARK_IMAGE> を手順 1 でビルドした Spark イメージに置き換えます
  image: <SPARK_IMAGE>
  imagePullPolicy: IfNotPresent
  mainClass: org.apache.spark.examples.SparkPi
  mainApplicationFile: local:///opt/spark/examples/jars/spark-examples_2.12-3.5.4.jar
  arguments:
  - "5000"
  sparkVersion: 3.5.4
  # 指定された ConfigMap からログ構成ファイル log4j2.properties を読み取ります
  sparkConfigMap: spark-log-conf
  sparkConf:
    # イベントログ
    spark.eventLog.enabled: "true"
    spark.eventLog.dir: file:///mnt/nas/spark/event-logs
  driver:
    cores: 1
    coreLimit: 1200m
    memory: 512m
    template:
      spec:
        containers:
        - name: spark-kubernetes-driver
          volumeMounts:
          - name: nas
            subPath: spark/event-logs
            mountPath: /mnt/nas/spark/event-logs
        serviceAccount: spark-operator-spark
        volumes:
        - name: nas
          persistentVolumeClaim:
            claimName: nas-pvc
  executor:
    instances: 1
    cores: 1
    coreLimit: 1200m
    memory: 512m
    template:
      spec:
        containers:
        - name: spark-kubernetes-executor
  restartPolicy:
    type: Never

「Celeborn を使用して Spark ジョブの RSS を有効にする」を参照して、クラスタに ack-celeborn コンポーネントをデプロイします。その後、以下のコードに基づいて Spark ジョブを送信し、Celeborn を RSS として使用できます。

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: spark-pagerank
  namespace: spark
spec:
  type: Scala
  mode: cluster
  # <SPARK_IMAGE> を Spark イメージに置き換えます
  image: <SPARK_IMAGE>
  imagePullPolicy: IfNotPresent
  mainClass: org.apache.spark.examples.SparkPageRank
  mainApplicationFile: local:///opt/spark/examples/jars/spark-examples_2.12-3.5.4.jar
  arguments:
  # 入力テストデータセットを指定します。<OSS_BUCKET> を OSS バケット名に置き換えます
  - oss://<OSS_BUCKET>/data/pagerank_dataset.txt
  # 反復回数
  - "10"
  sparkVersion: 3.5.4
  hadoopConf:
    # OSS への Spark ジョブアクセスを構成します
    fs.oss.impl: org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem
    # <OSS_ENDPOINT> を OSS アクセスエンドポイントに置き換えます。たとえば、北京の OSS の内部ネットワークアクセスエンドポイントは oss-cn-beijing-internal.aliyuncs.com です
    fs.oss.endpoint: <OSS_ENDPOINT>
    fs.oss.credentials.provider: com.aliyun.oss.common.auth.EnvironmentVariableCredentialsProvider
  # 指定された ConfigMap からログ構成ファイル log4j2.properties を読み取ります
  sparkConfigMap: spark-log-conf
  sparkConf:
    # イベントログ
    spark.eventLog.enabled: "true"
    spark.eventLog.dir: file:///mnt/nas/spark/event-logs
    
    # Celeborn 関連の構成
    spark.shuffle.manager: org.apache.spark.shuffle.celeborn.SparkShuffleManager
    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
    spark.celeborn.client.spark.shuffle.writer: hash
    spark.celeborn.client.push.replicate.enabled: "false"
    spark.sql.adaptive.localShuffleReader.enabled: "false"
    spark.sql.adaptive.enabled: "true"
    spark.sql.adaptive.skewJoin.enabled: "true"
    spark.shuffle.sort.io.plugin.class: org.apache.spark.shuffle.celeborn.CelebornShuffleDataIO
    spark.dynamicAllocation.shuffleTracking.enabled: "false"
    spark.executor.userClassPathFirst: "false"
  driver:
    cores: 1
    coreLimit: "1"
    memory: 4g
    template:
      spec:
        containers:
        - name: spark-kubernetes-driver
          envFrom:
          # 指定された Secret から環境変数を読み取ります
          - secretRef:
              name: spark-oss-secret
          volumeMounts:
          - name: nas
            subPath: spark/event-logs
            mountPath: /mnt/nas/spark/event-logs
        volumes:
        - name: nas
          persistentVolumeClaim:
            claimName: nas-pvc
        serviceAccount: spark-operator-spark
  executor:
    instances: 2
    cores: 1
    coreLimit: "1"
    memory: 4g
    template:
      spec:
        containers:
        - name: spark-kubernetes-executor
          envFrom:
          # 指定された Secret から環境変数を読み取ります
          - secretRef:
              name: spark-oss-secret
  restartPolicy:
    type: Never

次の例では、弾力性戦略をカスタマイズします。spark namespace の Spark Operator によって起動された Pod の場合、ECS リソースが優先され、最大 10 個の Pod を ECS インスタンスにスケジュールできます。ECS リソースが不足している場合は、Elastic Container Instance が使用され、最大 10 個の Pod を Elastic Container Instance にスケジュールできます。

apiVersion: scheduling.alibabacloud.com/v1alpha1
kind: ResourcePolicy
metadata:
  name: spark
  namespace: spark
spec:
  # ラベルセレクタを介して Pod に適用されるスケジューリング戦略を指定します
  selector:
    # たとえば、Spark Operator を介して送信された Pod に適用されるスケジューリング戦略を指定します
    sparkoperator.k8s.io/launched-by-spark-operator: "true"
  strategy: prefer
  # スケジューリングユニットの構成
  # スケールアウト中は、スケジューリングユニットの順序でスケールアウトが実行されます。スケールイン中は、スケジューリングユニットの逆順でスケールインが実行されます。
  units:
  # 最初のスケジューリングユニットは ECS リソースを使用し、最大 10 個の Pod を ECS インスタンスにスケジュールできます
  - resource: ecs
    max: 10
    # スケジューラはラベル情報を Pod に更新します
    podLabels:
      # これは Pod に更新されない特別なラベルです
      k8s.aliyun.com/resource-policy-wait-for-ecs-scaling: "true"
    # ECS リソースを使用する場合、ノードセレクタを介してスケジュール可能なノードを指定できます
    nodeSelector:
      # たとえば、従量課金タイプの ECS ノードを選択します
      node.alibabacloud.com/instance-charge-type: PostPaid
  # 2 番目のスケジューリングユニットは ECI リソースを使用し、最大 10 個の Pod を ECI インスタンスにスケジュールできます
  - resource: eci
    max: 10
  # Pod の数をカウントするときに、ResourcePolicy が作成される前にすでにスケジュールされていた Pod を無視します
  ignorePreviousPod: false
  # Pod の数をカウントするときに、Terminating 状態の Pod を無視します
  ignoreTerminatingPod: true
  # プリエンプション戦略
  # BeforeNextUnit は、各ユニットのスケジューリングが失敗した場合にスケジューラがプリエンプションを試みることを示します
  # AfterAllUnits は、最後のユニットのスケジューリングが失敗した場合にのみ ResourcePolicy がプリエンプションを試みることを示します
  preemptPolicy: AfterAllUnits
  # どのような状況で Pod が後続のユニットのリソースを使用できるかを指定します
  whenTryNextUnits:
    # 次の 2 つの条件のいずれかが満たされた場合、後続のユニットのリソースの使用が許可されます
    # 1. 現在のユニットの Max が設定されており、そのユニットの Pod の数が設定された Max 値以上である。
    # 2. 現在のユニットの Max が設定されておらず、そのユニットの PodLabels にラベル k8s.aliyun.com/resource-policy-wait-for-ecs-scaling: "true" が含まれており、タイムアウトまで待機した後。
    policy: TimeoutOrExceedMax
    # ポリシーが TimeoutOrExceedMax の場合、現在のユニットのリソースが Pod のスケジュールに不十分な場合、現在のユニットで待機し、最大待機時間はタイムアウトです。
    # この戦略は、自動スケーリングおよび ECI と組み合わせて使用​​して、ノードプール自動スケーリングを優先し、タイムアウト後に自動的に ECI を使用する効果を実現できます。
    timeout: 30s

このジョブの例では、Celeborn RSS と組み合わせて動的リソース割り当てをさらに構成します。

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: spark-pagerank
  namespace: spark
spec:
  type: Scala
  mode: cluster
  # <SPARK_IMAGE> を Spark イメージに置き換えます
  image: <SPARK_IMAGE>
  imagePullPolicy: IfNotPresent
  mainClass: org.apache.spark.examples.SparkPageRank
  mainApplicationFile: local:///opt/spark/examples/jars/spark-examples_2.12-3.5.4.jar
  arguments:
  # 入力テストデータセットを指定します。<OSS_BUCKET> を OSS バケット名に置き換えます
  - oss://<OSS_BUCKET>/data/pagerank_dataset.txt
  # 反復回数
  - "10"
  sparkVersion: 3.5.4
  hadoopConf:
    # OSS への Spark ジョブアクセスを構成します
    fs.oss.impl: org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem
    # <OSS_ENDPOINT> を OSS アクセスエンドポイントに置き換えます。たとえば、北京の OSS の内部ネットワークアクセスエンドポイントは oss-cn-beijing-internal.aliyuncs.com です
    fs.oss.endpoint: <OSS_ENDPOINT>
    fs.oss.credentials.provider: com.aliyun.oss.common.auth.EnvironmentVariableCredentialsProvider
  # 指定された ConfigMap からログ構成ファイル log4j2.properties を読み取ります
  sparkConfigMap: spark-log-conf
  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"
    # executor 数の初期値。
    spark.dynamicAllocation.initialExecutors: "3"
    # executor の最小数。
    spark.dynamicAllocation.minExecutors: "0"
    # executor の最大数。
    spark.dynamicAllocation.maxExecutors: "10"
    # executor のアイドルタイムアウト期間。タイムアウト期間を超えると、executor は解放されます。
    spark.dynamicAllocation.executorIdleTimeout: 60s
    # データブロックをキャッシュした executor のアイドルタイムアウト期間。タイムアウト期間を超えると、executor は解放されます。デフォルト値は無限大で、データブロックが解放されないことを指定します。
    # spark.dynamicAllocation.cachedExecutorIdleTimeout:
    # ジョブが指定されたスケジュール時間を超えると、追加の executor がリクエストされます。
    spark.dynamicAllocation.schedulerBacklogTimeout: 1s
    # 各時間間隔の後、後続のバッチの executor がリクエストされます。
    spark.dynamicAllocation.sustainedSchedulerBacklogTimeout: 1s
  driver:
    cores: 1
    coreLimit: "1"
    memory: 4g
    template:
      spec:
        containers:
        - name: spark-kubernetes-driver
          envFrom:
          # 指定された Secret から環境変数を読み取ります
          - secretRef:
              name: spark-oss-secret
          volumeMounts:
          - name: nas
            subPath: spark/event-logs
            mountPath: /mnt/nas/spark/event-logs
        volumes:
        - name: nas
          persistentVolumeClaim:
            claimName: nas-pvc
        serviceAccount: spark-operator-spark
  executor:
    cores: 1
    coreLimit: "1"
    memory: 4g
    template:
      spec:
        containers:
        - name: spark-kubernetes-executor
          envFrom:
          # 指定された Secret から環境変数を読み取ります
          - secretRef:
              name: spark-oss-secret
  restartPolicy:
    type: Never