Kubernetes 上での Spark 実行

• セキュリティ
  • ユーザーID
  • ボリュームマウント
• 前提条件
• 動作方法
• Kubernetes へのアプリケーションの送信
  • Docker イメージ
  • クラスタモード
  • クライアントモード
    • クライアントモードのネットワーク
    • クライアントモードのエグゼキュータポッドのガベージコレクション
    • 認証パラメータ
  • IPv4 と IPv6
  • 依存関係管理
  • シークレット管理
  • ポッドテンプレート
  • Kubernetes ボリュームの使用
    • PVC ベースのエグゼキュータポッド割り当て
  • ローカルストレージ
    • ローカルストレージとしてのRAMの使用
  • イントロスペクションとデバッグ
    • ログへのアクセス
    • ドライバUIへのアクセス
    • デバッグ
  • Kubernetes の機能
    • 設定ファイル
    • コンテキスト
    • 名前空間
    • RBAC
  • Spark アプリケーション管理
  • 今後の作業
• 設定
  • Spark プロパティ
  • ポッドテンプレートプロパティ
  • ポッドメタデータ
  • ポッド仕様
  • コンテナ仕様
  • リソース割り当てと設定の概要
  • リソースレベルスケジューリングの概要
    • 優先順位スケジューリング
    • Kubernetes 上の Spark 用のカスタマイズされた Kubernetes スケジューラ
    • Kubernetes 上の Spark 用のカスタマイズされたスケジューラとしての Volcano の使用
      • 前提条件
      • ビルド
      • 使用方法
      • Volcano 機能ステップ
      • Volcano PodGroup テンプレート
    • Kubernetes 上の Spark 用のカスタマイズされたスケジューラとしての Apache YuniKorn の使用
      • 前提条件
      • はじめに
  • ステージレベルスケジューリングの概要

Spark は、Kubernetes によって管理されるクラスタ上で実行できます。この機能は、Spark に追加されたネイティブの Kubernetes スケジューラを利用しています。

セキュリティ

認証などのセキュリティ機能は、デフォルトでは有効になっていません。インターネットまたは信頼できないネットワークに公開されているクラスタをデプロイする場合は、クラスタへのアクセスを保護して、不正なアプリケーションがクラスタで実行されないようにすることが重要です。Spark を実行する前に、Spark セキュリティ とこのドキュメントの特定のセキュリティセクションを参照してください。

ユーザーID

プロジェクト提供の Dockerfile からビルドされたイメージには、185 のデフォルト UID を持つデフォルトの USER ディレクティブが含まれています。これは、結果として得られるイメージが、コンテナ内でこの UID として Spark プロセスを実行することを意味します。セキュリティを重視したデプロイメントでは、目的の非特権 UID と GID を指定する USER ディレクティブを指定したカスタムイメージを提供することを検討する必要があります。結果の UID には、Spark 実行可能ファイルを動作させるために、ルートグループを補足グループに含める必要があります。提供されている docker-image-tool.sh スクリプトで独自のイメージをビルドするユーザーは、-u <uid> オプションを使用して目的の UID を指定できます。

あるいは、ポッドテンプレート 機能を使用して、Spark が送信するポッドに runAsUser を含む セキュリティコンテキスト を追加できます。これは、イメージ自体の USER ディレクティブをオーバーライドするために使用できます。ただし、これはユーザーの協力が必要となるため、共有環境には適さないソリューションとなる可能性があることに注意してください。クラスタ管理者は、ポッドが実行できるユーザーを制限する場合、Pod セキュリティポリシー を使用する必要があります。

ボリュームマウント

このドキュメントの後半にある Kubernetes ボリュームの使用 の項で説明されているように、K8S 上の Spark は、特定の種類のボリュームをドライバとエグゼキュータポッドにマウントできるようにする設定オプションを提供しています。特に、Kubernetes のドキュメントで説明されているように、既知のセキュリティ脆弱性を持つ hostPath ボリュームを許可しています。

クラスタ管理者は、Pod セキュリティポリシー を使用して、hostPath ボリュームのマウント機能を環境に合わせて適切に制限する必要があります。

前提条件

• kubectl を使用してアクセスが設定されているバージョン 1.24 以上の実行中の Kubernetes クラスタ。動作中の Kubernetes クラスタがない場合は、minikube を使用してローカルマシンにテストクラスタをセットアップできます。
  • DNS アドオンが有効になっている minikube の最新リリースを使用することをお勧めします。
  • デフォルトの minikube 設定では、Spark アプリケーションを実行するには不十分であることに注意してください。単一のエグゼキュータを持つ単純な Spark アプリケーションを起動するには、3 つの CPU と 4GB のメモリを推奨します。
  • Spark 環境の kubernetes-client ライブラリ のバージョンと、Kubernetes クラスタのバージョンとの互換性を確認してください。
• クラスタ内の ポッド をリスト、作成、編集、削除するための適切な権限が必要です。これらのリソースをリストできることを確認するには、kubectl auth can-i <list|create|edit|delete> pods を実行します。
  • ドライバポッドで使用されるサービスアカウントの資格情報は、ポッド、サービス、ConfigMap を作成できる必要があります。
• クラスタに Kubernetes DNS が設定されている必要があります。

動作方法

spark-submit を直接使用して、Spark アプリケーションを Kubernetes クラスタに送信できます。送信メカニズムは次のとおりです。

• Spark は、Kubernetes ポッド 内で実行される Spark ドライバを作成します。
• ドライバは、Kubernetes ポッド内でも実行され、それらに接続し、アプリケーションコードを実行するエグゼキュータを作成します。
• アプリケーションが完了すると、エグゼキュータポッドは終了してクリーンアップされますが、ドライバポッドはログを保持し、最終的にガベージコレクションされるか手動でクリーンアップされるまで、Kubernetes API で「完了」状態のままになります。

完了状態では、ドライバポッドは *計算* リソースまたはメモリリソースを使用しないことに注意してください。

ドライバとエグゼキュータポッドのスケジューリングは、Kubernetes によって処理されます。Kubernetes API への通信は fabric8 を介して行われます。設定プロパティを使用して、ノードセレクタ を介して使用可能なノードのサブセットにドライバとエグゼキュータポッドをスケジュールできます。ノード/ポッドアフィニティ などの高度なスケジューリングヒントは、今後のリリースで使用できるようになります。

Kubernetes へのアプリケーションの送信

Docker イメージ

Kubernetes では、ポッド内のコンテナにデプロイできるイメージをユーザーが提供する必要があります。イメージは、Kubernetes がサポートするコンテナランタイム環境で実行されるようにビルドされます。Docker は、Kubernetes で頻繁に使用されるコンテナランタイム環境です。Spark（バージョン 2.3 以降）には、この目的で使用できる、または個々のアプリケーションのニーズに合わせてカスタマイズできる Dockerfile が付属しています。これは、kubernetes/dockerfiles/ ディレクトリにあります。

Spark には、Kubernetes バックエンドで使用される Docker イメージをビルドおよび公開するために使用できる bin/docker-image-tool.sh スクリプトも付属しています。

使用方法の例を次に示します。

$ ./bin/docker-image-tool.sh -r <repo> -t my-tag build
$ ./bin/docker-image-tool.sh -r <repo> -t my-tag push

これにより、プロジェクトが提供するデフォルトの Dockerfiles を使用してビルドが行われます。カスタム Dockerfiles の提供など、このツールの動作をカスタマイズするための利用可能なオプションの詳細については、-h フラグを付けて実行してください。

デフォルトでは、bin/docker-image-tool.sh は JVM ジョブを実行するための Docker イメージをビルドします。追加の言語バインディング Docker イメージをビルドするには、オプトインする必要があります。

使用方法の例を次に示します。

# To build additional PySpark docker image
$ ./bin/docker-image-tool.sh -r <repo> -t my-tag -p ./kubernetes/dockerfiles/spark/bindings/python/Dockerfile build

# To build additional SparkR docker image
$ ./bin/docker-image-tool.sh -r <repo> -t my-tag -R ./kubernetes/dockerfiles/spark/bindings/R/Dockerfile build

Apache Spark Docker イメージ（apache/spark:<version> など）を直接使用することもできます。

クラスタモード

クラスタモードで Spark Pi を起動するには、

$ ./bin/spark-submit \
    --master k8s://https://<k8s-apiserver-host>:<k8s-apiserver-port> \
    --deploy-mode cluster \
    --name spark-pi \
    --class org.apache.spark.examples.SparkPi \
    --conf spark.executor.instances=5 \
    --conf spark.kubernetes.container.image=<spark-image> \
    local:///path/to/examples.jar

spark-submit に --master コマンドライン引数を渡すか、アプリケーションの設定で spark.master を設定することで指定された Spark マスターは、k8s://<api_server_host>:<k8s-apiserver-port> 形式の URL でなければなりません。HTTPS ポート 443 であっても、ポートを常に指定する必要があります。k8s:// をマスター文字列の前に付けることで、Spark アプリケーションは Kubernetes クラスタで起動され、api_server_url で API サーバーに接続されます。URL に HTTP プロトコルが指定されていない場合、デフォルトで https になります。たとえば、マスターを k8s://example.com:443 に設定することは、k8s://https://example.com:443 に設定することと同じですが、異なるポートで TLS なしで接続するには、マスターを k8s://http://example.com:8080 に設定します。

Kubernetes モードでは、spark.app.name または spark-submit の --name 引数で指定された Spark アプリケーション名は、デフォルトでドライバやエグゼキュータなどの作成された Kubernetes リソースの名前付けに使用されます。したがって、アプリケーション名は、小文字の英数字、-、. で構成され、英数字で始まる必要があります。

Kubernetes クラスタが設定されている場合、apiserver URL を検出する1つの方法は、kubectl cluster-info を実行することです。

$ kubectl cluster-info
Kubernetes master is running at http://127.0.0.1:6443

上記の例では、`spark-submit` に `--master k8s://http://127.0.0.1:6443` を引数として指定することで、特定のKubernetesクラスタを使用できます。さらに、認証プロキシである`kubectl proxy` を使用してKubernetes APIと通信することも可能です。

ローカルプロキシは以下のように開始できます。

$ kubectl proxy

ローカルプロキシがlocalhost:8001で実行されている場合、`--master k8s://http://127.0.0.1:8001` を`spark-submit` の引数として使用できます。最後に、上記の例では、スキームが`local://`である特定のURIを持つjarを指定していることに注意してください。このURIは、Dockerイメージに既に存在するサンプルjarの場所です。

クライアントモード

Spark 2.4.0以降、クライアントモードでKubernetes上でSparkアプリケーションを実行できます。アプリケーションがクライアントモードで実行されると、ドライバはpod内または物理ホスト上で実行できます。クライアントモードでアプリケーションを実行する場合は、次の要素を考慮することをお勧めします。

クライアントモードのネットワーク

Sparkエグゼキュータは、Sparkエグゼキュータからルーティング可能なホスト名とポートを介してSparkドライバに接続できる必要があります。Sparkがクライアントモードで動作するために必要な特定のネットワーク構成は、設定によって異なります。Kubernetes pod内でドライバを実行する場合は、ヘッドレスサービス を使用して、ドライバpodが安定したホスト名によってエグゼキュータからルーティングできるようにします。ヘッドレスサービスをデプロイする際には、サービスのラベルセレクタがドライバpodのみに一致し、他のpodには一致しないようにしてください。ドライバpodに十分に一意のラベルを割り当て、そのラベルをヘッドレスサービスのラベルセレクタで使用することをお勧めします。ドライバのホスト名を`spark.driver.host`、Sparkドライバのポートを`spark.driver.port`で指定します。

クライアントモードのエグゼキュータポッドのガベージコレクション

Sparkドライバをpod内で実行する場合は、`spark.kubernetes.driver.pod.name`をそのpodの名前に設定することを強くお勧めします。このプロパティが設定されると、SparkスケジューラはOwnerReferenceを使用してエグゼキュータpodをデプロイします。これにより、ドライバpodがクラスタから削除されると、アプリケーションのエグゼキュータpodもすべて削除されます。ドライバは、`spark.kubernetes.namespace`で指定された名前空間にある指定された名前のpodを探し、そのpodを指すOwnerReferenceが各エグゼキュータpodのOwnerReferencesリストに追加されます。実際にはそのドライバpodではないpodにOwnerReferenceを設定しないように注意してください。そうしないと、間違ったpodが削除されたときにエグゼキュータが早期に終了される可能性があります。

アプリケーションがpod内で実行されていない場合、またはアプリケーションが実際にpod内で実行されているときに`spark.kubernetes.driver.pod.name`が設定されていない場合は、アプリケーションが終了したときにエグゼキュータpodがクラスタから適切に削除されない可能性があることに注意してください。Sparkスケジューラはこれらのpodの削除を試みますが、何らかの理由でAPIサーバへのネットワークリクエストが失敗した場合、これらのpodはクラスタに残ります。エグゼキュータプロセスは、ドライバに到達できない場合に終了するため、アプリケーションの終了後、エグゼキュータpodはクラスタで計算リソース（CPUとメモリ）を消費しません。

`spark.kubernetes.executor.podNamePrefix`を使用して、エグゼキュータpod名を完全に制御できます。このプロパティを設定する場合は、同じ名前空間内のすべてのジョブで一意にすることを強くお勧めします。

認証パラメータ

クライアントモードでのKubernetes認証パラメータには、`spark.kubernetes.authenticate`という正確なプレフィックスを使用してください。

IPv4 と IPv6

3.4.0以降、SparkはIPv4/IPv6デュアルスタックネットワーク機能を介してIPv6のみの環境もサポートしています。この機能により、PodとサービスにIPv4とIPv6の両方のアドレスを割り当てることができます。K8sクラスタの機能に応じて、`spark.kubernetes.driver.service.ipFamilyPolicy`と`spark.kubernetes.driver.service.ipFamilies`は`SingleStack`、`PreferDualStack`、`RequireDualStack`のいずれか、および`IPv4`、`IPv6`、`IPv4,IPv6`、`IPv6,IPv4`のいずれかになります。デフォルトでは、Sparkは`spark.kubernetes.driver.service.ipFamilyPolicy=SingleStack`と`spark.kubernetes.driver.service.ipFamilies=IPv4`を使用します。

`IPv6`のみを使用するには、次のコマンドでジョブを送信します。

...
    --conf spark.kubernetes.driver.service.ipFamilies=IPv6 \

`DualStack`環境では、`IPv6`を使用するために、JVMには`java.net.preferIPv6Addresses=true`、Pythonには`SPARK_PREFER_IPV6=true`も追加で必要となる場合があります。

依存関係管理

アプリケーションの依存関係がすべてHDFSやHTTPサーバなどのリモートロケーションにホストされている場合、それらは適切なリモートURIで参照できます。また、アプリケーションの依存関係は、カスタムビルドのDockerイメージに事前にマウントできます。これらの依存関係は、`local://` URIで参照するか、Dockerfileで`SPARK_EXTRA_CLASSPATH`環境変数を設定することで、クラスパスに追加できます。`local://`スキームは、`spark-submit`でカスタムビルドのDockerイメージ内の依存関係を参照する場合にも必要です。提出クライアントのローカルファイルシステムからの依存関係は、`file://`スキームを使用するか、スキームなしで（フルパスを使用して）サポートしており、宛先はHadoop互換のファイルシステムである必要があります。S3を使用した典型的な例は、次のオプションを渡すことです。

...
--packages org.apache.hadoop:hadoop-aws:3.2.2
--conf spark.kubernetes.file.upload.path=s3a://<s3-bucket>/path
--conf spark.hadoop.fs.s3a.access.key=...
--conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem
--conf spark.hadoop.fs.s3a.fast.upload=true
--conf spark.hadoop.fs.s3a.secret.key=....
--conf spark.driver.extraJavaOptions=-Divy.cache.dir=/tmp -Divy.home=/tmp
file:///full/path/to/app.jar

アプリのjarファイルはS3にアップロードされ、ドライバが起動されるとドライバpodにダウンロードされ、そのクラスパスに追加されます。Sparkは、並行して実行されているSparkアプリとの競合を回避するために、ランダムな名前のアップロードパスの下にサブディレクトリを生成します。ユーザーは必要に応じて作成されたサブディレクトリを管理できます。

クライアントスキームは、アプリケーションjar、およびプロパティ`spark.jars`、`spark.files`、`spark.archives`で指定された依存関係でサポートされています。

重要：すべてのクライアント側の依存関係は、フラットなディレクトリ構造で指定されたパスにアップロードされるため、ファイル名が重複するとファイルが上書きされます。また、派生したk8sイメージのデフォルトのivyディレクトリに必要なアクセス権限があること、または上記のように設定を変更してください。後者は、クラスタモードで`--packages`を使用する場合にも重要です。

シークレット管理

Kubernetes Secrets を使用して、Sparkアプリケーションが保護されたサービスにアクセスするための資格情報を提供できます。ユーザー指定のシークレットをドライバコンテナにマウントするには、`spark.kubernetes.driver.secrets.[SecretName]=<mount path>`形式の構成プロパティを使用できます。同様に、`spark.kubernetes.executor.secrets.[SecretName]=<mount path>`形式の構成プロパティを使用して、ユーザー指定のシークレットをエグゼキュータコンテナにマウントできます。マウントするシークレットは、ドライバとエグゼキュータpodと同じ名前空間に存在すると仮定されています。たとえば、`spark-secret`という名前のシークレットをドライバとエグゼキュータコンテナの両方で`/etc/secrets`パスにマウントするには、`spark-submit`コマンドに次のオプションを追加します。

--conf spark.kubernetes.driver.secrets.spark-secret=/etc/secrets
--conf spark.kubernetes.executor.secrets.spark-secret=/etc/secrets

環境変数を使用してシークレットを使用するには、`spark-submit`コマンドに次のオプションを使用します。

--conf spark.kubernetes.driver.secretKeyRef.ENV_NAME=name:key
--conf spark.kubernetes.executor.secretKeyRef.ENV_NAME=name:key

ポッドテンプレート

Kubernetesでは、テンプレートファイルからpodを定義できます。Sparkユーザーは同様に、テンプレートファイルを使用して、Spark構成がサポートしていないドライバまたはエグゼキュータpod構成を定義できます。これを行うには、`spark-submit`プロセスからアクセス可能なファイルを参照するように、`spark.kubernetes.driver.podTemplateFile`と`spark.kubernetes.executor.podTemplateFile`のSparkプロパティを指定します。

--conf spark.kubernetes.driver.podTemplateFile=s3a://bucket/driver.yml
--conf spark.kubernetes.executor.podTemplateFile=s3a://bucket/executor.yml

ドライバpodがエグゼキュータpodテンプレートファイルにアクセスできるようにするために、ファイルは作成時にドライバpodのボリュームに自動的にマウントされます。Sparkはこれらのテンプレートファイルをアンマーシャリングした後、検証を行わず、検証にはKubernetes APIサーバに依存します。

Sparkは特定のpod構成について独自の考えを持っているため、podテンプレート内の値は常にSparkによって上書きされることに注意することが重要です。したがって、この機能のユーザーは、podテンプレートファイルを指定しても、pod構築プロセス中に空のpodではなくテンプレートpodから開始することだけをSparkに許可することに注意してください。詳細については、Sparkによって上書きされるpodテンプレート値の完全なリストを参照してください。

Podテンプレートファイルは、複数のコンテナも定義できます。このような場合、`spark.kubernetes.driver.podTemplateContainerName`と`spark.kubernetes.executor.podTemplateContainerName`のSparkプロパティを使用して、ドライバまたはエグゼキュータのベースとして使用するコンテナを示すことができます。指定されていない場合、またはコンテナ名が有効でない場合、Sparkはリスト内の最初のコンテナがドライバまたはエグゼキュータコンテナであると仮定します。

Kubernetes ボリュームの使用

ユーザーは、次のタイプのKubernetes ボリュームをドライバとエグゼキュータpodにマウントできます。

• hostPath：ホストノードのファイルシステムからpodにファイルまたはディレクトリをマウントします。
• emptyDir：podがノードに割り当てられたときに作成される、最初は空のボリューム。
• nfs：既存のNFS（Network File System）をpodにマウントします。
• persistentVolumeClaim：`PersistentVolume`をpodにマウントします。

**注記：** ボリュームマウントに関連するセキュリティの問題については、このドキュメントのセキュリティセクションを参照してください。

上記のいずれかのタイプのボリュームをドライバpodにマウントするには、次の構成プロパティを使用します。

--conf spark.kubernetes.driver.volumes.[VolumeType].[VolumeName].mount.path=<mount path>
--conf spark.kubernetes.driver.volumes.[VolumeType].[VolumeName].mount.readOnly=<true|false>
--conf spark.kubernetes.driver.volumes.[VolumeType].[VolumeName].mount.subPath=<mount subPath>

具体的には、`VolumeType`は`hostPath`、`emptyDir`、`nfs`、`persistentVolumeClaim`のいずれかの値になります。`VolumeName`は、pod仕様の`volumes`フィールドでボリュームに使用したい名前です。

サポートされている各タイプのボリュームには、いくつかの特定の構成オプションがあり、次の形式の構成プロパティを使用して指定できます。

spark.kubernetes.driver.volumes.[VolumeType].[VolumeName].options.[OptionName]=<value>

たとえば、ボリューム名`images`の`nfs`のサーバとパスは、次のプロパティを使用して指定できます。

spark.kubernetes.driver.volumes.nfs.images.options.server=example.com
spark.kubernetes.driver.volumes.nfs.images.options.path=/data

そして、ボリューム名checkpointpvcを持つpersistentVolumeClaimのクレーム名は、以下のプロパティを使用して指定できます。

spark.kubernetes.driver.volumes.persistentVolumeClaim.checkpointpvc.options.claimName=check-point-pvc-claim

エグゼキュータポッドにボリュームをマウントするための構成プロパティは、spark.kubernetes.driver.ではなくspark.kubernetes.executor.というプレフィックスを使用します。

例えば、OnDemandをクレーム名として使用し、storageClassとsizeLimitオプションを以下のように使用することで、エグゼキュータごとに動的に作成されたパーシステントボリュームクレームをマウントできます。これは動的割り当ての場合に役立ちます。

spark.kubernetes.executor.volumes.persistentVolumeClaim.data.options.claimName=OnDemand
spark.kubernetes.executor.volumes.persistentVolumeClaim.data.options.storageClass=gp
spark.kubernetes.executor.volumes.persistentVolumeClaim.data.options.sizeLimit=500Gi
spark.kubernetes.executor.volumes.persistentVolumeClaim.data.mount.path=/data
spark.kubernetes.executor.volumes.persistentVolumeClaim.data.mount.readOnly=false

サポートされている各ボリュームの種類で使用可能なオプションの完全なリストについては、以下のSparkプロパティセクションを参照してください。

PVC ベースのエグゼキュータポッド割り当て

ディスクは重要なリソースタイプの1つであるため、Sparkドライバは一連の構成を通じてきめ細かい制御を提供します。例えば、デフォルトでは、オンデマンドPVCはエグゼキュータが所有し、PVCのライフサイクルは所有者であるエグゼキュータと密接に関連付けられています。しかし、以下のオプションを使用することで、オンデマンドPVCをドライバが所有し、Sparkジョブのライフタイム中に他のエグゼキュータで再利用できます。これにより、PVCの作成と削除のオーバーヘッドを削減できます。

spark.kubernetes.driver.ownPersistentVolumeClaim=true
spark.kubernetes.driver.reusePersistentVolumeClaim=true

さらに、Spark 3.4以降、SparkドライバはPVC指向のエグゼキュータ割り当てを実行できるようになりました。これは、ジョブが持つことができる作成済みPVCの総数をSparkがカウントし、ドライバが最大数のPVCを所有している場合、新しいエグゼキュータの作成を保留することを意味します。これは、既存のPVCを1つのエグゼキュータから別のエグゼキュータへの移行に役立ちます。

spark.kubernetes.driver.waitToReusePersistentVolumeClaim=true

ローカルストレージ

Sparkは、シャッフルやその他の操作中にデータを書き出すためにボリュームの使用をサポートしています。ボリュームをローカルストレージとして使用するには、ボリュームの名前をspark-local-dir-で始める必要があります。例えば

--conf spark.kubernetes.driver.volumes.[VolumeType].spark-local-dir-[VolumeName].mount.path=<mount path>
--conf spark.kubernetes.driver.volumes.[VolumeType].spark-local-dir-[VolumeName].mount.readOnly=false

特に、ジョブでエグゼキュータで大規模なシャッフルとソート操作が必要な場合は、パーシステントボリュームクレームを使用できます。

spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-1.options.claimName=OnDemand
spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-1.options.storageClass=gp
spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-1.options.sizeLimit=500Gi
spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-1.mount.path=/data
spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-1.mount.readOnly=false

組み込みのKubernetesLocalDiskShuffleDataIOプラグインを介してシャッフルデータ復旧機能を有効にするには、以下のものが必要です。spark.kubernetes.driver.waitToReusePersistentVolumeClaimも追加で有効にしたい場合があります。

spark.kubernetes.executor.volumes.persistentVolumeClaim.spark-local-dir-1.mount.path=/data/spark-x/executor-x
spark.shuffle.sort.io.plugin.class=org.apache.spark.shuffle.KubernetesLocalDiskShuffleDataIO

ローカルストレージとしてボリュームが設定されていない場合、Sparkはシャッフルやその他の操作中にディスクにデータを書き出すために一時的なスクラッチスペースを使用します。Kubernetesをリソースマネージャーとして使用する場合、ポッドはspark.local.dirまたは環境変数SPARK_LOCAL_DIRSにリストされている各ディレクトリにマウントされたemptyDirボリュームで作成されます。ディレクトリが明示的に指定されていない場合、デフォルトのディレクトリが作成され、適切に構成されます。

emptyDirボリュームはKubernetesの一時ストレージ機能を使用し、ポッドの寿命を超えては永続化されません。

ローカルストレージとしてのRAMの使用

emptyDirボリュームはデフォルトでノードのバッキングストレージを一時ストレージに使用しますが、この動作は一部のコンピューティング環境には適さない場合があります。例えば、ネットワーク経由でマウントされたリモートストレージを持つディスクレスノードがある場合、多くのエグゼキュータがこのリモートストレージへのIOを実行すると、実際のパフォーマンスが低下する可能性があります。

この場合、構成でspark.kubernetes.local.dirs.tmpfs=trueを設定すると、emptyDirボリュームがtmpfsつまりRAMバックされたボリュームとして構成されます。このように構成されている場合、Sparkのローカルストレージの使用量はポッドのメモリ使用量にカウントされるため、必要に応じてspark.{driver,executor}.memoryOverheadFactorの値を増やすことでメモリ要求を増やすことをお勧めします。

イントロスペクションとデバッグ

これらは、実行中/完了したSparkアプリケーションを調査し、進捗状況を監視し、アクションを実行できるさまざまな方法です。

ログへのアクセス

ログは、Kubernetes APIとkubectl CLIを使用してアクセスできます。Sparkアプリケーションが実行されている場合、アプリケーションからログをストリーミングできます。

$ kubectl -n=<namespace> logs -f <driver-pod-name>

同じログは、クラスタにインストールされている場合はKubernetesダッシュボードからもアクセスできます。

ドライバUIへのアクセス

任意のアプリケーションに関連付けられたUIには、kubectl port-forwardを使用してローカルでアクセスできます。

$ kubectl port-forward <driver-pod-name> 4040:4040

その後、SparkドライバUIにはhttp://localhost:4040でアクセスできます。

デバッグ

いくつかの種類の障害が発生する可能性があります。Kubernetes APIサーバーがspark-submitから行われたリクエストを拒否した場合、または別の理由で接続が拒否された場合、送信ロジックは発生したエラーを示す必要があります。ただし、アプリケーションの実行中にエラーが発生した場合、多くの場合、調査する最良の方法はKubernetes CLIを使用することです。

ドライバポッドに関するスケジューリングの決定に関する基本的な情報を取得するには、以下を実行します。

$ kubectl describe pod <spark-driver-pod>

ポッドでランタイムエラーが発生した場合は、以下を使用してステータスをさらに調査できます。

$ kubectl logs <spark-driver-pod>

失敗したエグゼキュータポッドのステータスとログは、同様の方法で確認できます。最後に、ドライバポッドを削除すると、すべてのエグゼキュータ、関連サービスなどを含む、すべてのSparkアプリケーションがクリーンアップされます。ドライバポッドは、SparkアプリケーションのKubernetes表現と考えることができます。

Kubernetes の機能

設定ファイル

Kubernetesの設定ファイルは、通常、ホームディレクトリの.kube/configまたはKUBECONFIG環境変数で指定された場所に存在します。Spark on Kubernetesは、このファイルを使用して、Kubernetesクラスタとのやり取りに使用されるKubernetesクライアントの初期自動構成を試みます。代替認証方法の使用など、クライアント構成をさらにカスタマイズできるさまざまなSpark構成プロパティが提供されています。

コンテキスト

Kubernetesの設定ファイルには、複数のクラスタやユーザーID間を切り替えることができる複数のコンテキストを含めることができます。デフォルトでは、Spark on Kubernetesは、Kubernetesクライアントの初期自動構成を行う際に、現在のコンテキスト（kubectl config current-contextを実行して確認できます）を使用します。

代替コンテキストを使用するには、ユーザーはSpark構成プロパティspark.kubernetes.context（例：spark.kubernetes.context=minikube）で目的のコンテキストを指定できます。

名前空間

Kubernetesにはネームスペースの概念があります。ネームスペースは、複数のユーザー間でクラスタリソースを分割する方法です（リソースクォータを介して）。Spark on Kubernetesは、Sparkアプリケーションを起動するためにネームスペースを使用できます。これはspark.kubernetes.namespace構成を使用して利用できます。

KubernetesではResourceQuotaを使用して、個々のネームスペースのリソース、オブジェクトの数などに制限を設定できます。管理者は、ネームスペースとResourceQuotaを組み合わせて使用することで、Sparkアプリケーションを実行するKubernetesクラスタでの共有とリソース割り当てを制御できます。

RBAC

RBACが有効になっているKubernetesクラスタでは、ユーザーは、さまざまなSpark on KubernetesコンポーネントがKubernetes APIサーバーにアクセスするために使用するKubernetes RBACロールとサービスアカウントを構成できます。

Sparkドライバポッドは、Kubernetesサービスアカウントを使用してKubernetes APIサーバーにアクセスし、エグゼキュータポッドを作成および監視します。ドライバポッドで使用されるサービスアカウントには、ドライバが作業を実行するために適切な権限が必要です。具体的には、少なくとも、サービスアカウントには、ドライバポッドがポッドとサービスを作成できるRoleまたはClusterRoleを付与する必要があります。デフォルトでは、ポッドの作成時にサービスアカウントが指定されていない場合、ドライバポッドにはspark.kubernetes.namespaceで指定されたネームスペースのdefaultサービスアカウントが自動的に割り当てられます。

Kubernetesのバージョンと設定によっては、このdefaultサービスアカウントに、デフォルトのKubernetes RBACポリシーの下でドライバポッドがポッドとサービスを作成できるロールがある場合とない場合があります。場合によっては、適切なロールが付与されたカスタムサービスアカウントを指定する必要がある場合があります。Spark on Kubernetesは、構成プロパティspark.kubernetes.authenticate.driver.serviceAccountName=<service account name>を使用してドライバポッドで使用されるカスタムサービスアカウントの指定をサポートしています。例えば、ドライバポッドでsparkサービスアカウントを使用するには、ユーザーはspark-submitコマンドに次のオプションを追加するだけです。

--conf spark.kubernetes.authenticate.driver.serviceAccountName=spark

カスタムサービスアカウントを作成するには、ユーザーはkubectl create serviceaccountコマンドを使用できます。例えば、次のコマンドはsparkという名前のサービスアカウントを作成します。

$ kubectl create serviceaccount spark

サービスアカウントにRoleまたはClusterRoleを付与するには、RoleBindingまたはClusterRoleBindingが必要です。RoleBindingまたはClusterRoleBindingを作成するには、ユーザーはkubectl create rolebinding（またはClusterRoleBindingの場合はclusterrolebinding）コマンドを使用できます。例えば、次のコマンドはdefaultネームスペースにedit ClusterRoleを作成し、それを上記で作成したsparkサービスアカウントに付与します。

$ kubectl create clusterrolebinding spark-role --clusterrole=edit --serviceaccount=default:spark --namespace=default

Roleは単一のネームスペース内のリソース（ポッドなど）へのアクセス権を付与するためにのみ使用できますが、ClusterRoleは、すべてのネームスペースにわたるクラスタスコープのリソース（ノードなど）とネームスペースのリソース（ポッドなど）へのアクセス権を付与するために使用できます。Spark on Kubernetesでは、ドライバは常に同じネームスペースにエグゼキュータポッドを作成するため、Roleで十分ですが、ユーザーは代わりにClusterRoleを使用できます。RBAC認証と、ポッドのKubernetesサービスアカウントの構成方法の詳細については、RBAC認証の使用とポッドのサービスアカウントの構成を参照してください。

Spark アプリケーション管理

Kubernetesは、クラスタモードでspark-submit CLIツールを介して、単純なアプリケーション管理を提供します。ユーザーは、ジョブの送信時に表示される送信IDを提供することで、ジョブを強制終了できます。送信IDの形式はnamespace:driver-pod-nameです。ユーザーがネームスペースを省略した場合、現在のk8sコンテキストに設定されているネームスペースが使用されます。例えば、ユーザーが次のように特定のネームスペースを設定した場合kubectl config set-context minikube --namespace=spark、sparkネームスペースがデフォルトで使用されます。一方、特定のコンテキストにネームスペースが追加されていない場合、すべてのネームスペースがデフォルトで考慮されます。つまり、ネームスペースに関係なく、指定された送信IDと一致するすべてのSparkアプリケーションに操作が影響します。さらに、アプリケーション管理用のspark-submitは、ドライバを送信するために使用されるのと同じバックエンドコードを使用するため、spark.kubernetes.contextなどの同じプロパティを再利用できます。

例えば

$ spark-submit --kill spark:spark-pi-1547948636094-driver --master k8s://https://192.168.2.8:8443

ユーザーは、--statusフラグを使用してアプリケーションのステータスをリストすることもできます。

$ spark-submit --status spark:spark-pi-1547948636094-driver --master  k8s://https://192.168.2.8:8443

どちらの操作もglobパターンをサポートしています。例えば、ユーザーは以下を実行できます。

$ spark-submit --kill spark:spark-pi* --master  k8s://https://192.168.2.8:8443

上記は、特定のプレフィックスを持つすべてのアプリケーションを強制終了します。

ユーザーは、spark.kubernetes.appKillPodDeletionGracePeriodプロパティを介して、ポッドの終了の猶予期間を指定できます。これは--confを使用して提供します（すべてのK8sポッドのデフォルト値は30秒です）。

今後の作業

現在開発中、もしくは開発予定のSpark on Kubernetes機能がいくつかあります。これらの機能は、将来のバージョンのspark-kubernetes統合に含まれる予定です。

これらの機能の一部を以下に示します。

• 外部シャッフルサービス
• ジョブキューとリソース管理

設定

Sparkの設定に関する情報は、設定ページを参照してください。以下の設定は、Spark on Kubernetesに固有のものです。

Spark プロパティ

ポッドテンプレートプロパティ

Sparkによって上書きされるポッド仕様の完全なリストについては、以下の表を参照してください。

ポッドメタデータ

ポッド仕様

コンテナ仕様

以下は、ドライバとエグゼキュータのコンテナに影響を与えます。ポッド仕様の他のすべてのコンテナは影響を受けません。

リソース割り当てと設定の概要

設定ページのカスタムリソースのスケジューリングと設定の概要セクションをよくお読みください。このセクションでは、リソーススケジューリングのKubernetes固有の側面についてのみ説明しています。

ユーザーは、リソースが利用可能になるようにKubernetesクラスタを適切に設定し、理想的にはコンテナごとにリソースを分離して、リソースが複数のコンテナ間で共有されないようにする責任があります。リソースが分離されていない場合、ユーザーは、リソースがコンテナ間で共有されないようにするための検出スクリプトを作成する責任があります。カスタムリソースを使用してKubernetesを設定する方法の詳細については、Kubernetesのドキュメントを参照してください。

Sparkは、Kubernetesリソースタイプがvendor-domain/resourcetypeのKubernetesデバイスプラグイン形式に従う限り、Spark設定spark.{driver/executor}.resource.{resourceType}をKubernetes設定に変換する処理を自動的に行います。ユーザーは、spark.{driver/executor}.resource.{resourceType}.vendor設定を使用してベンダーを指定する必要があります。ポッドテンプレートを使用している場合は、明示的に何も追加する必要はありません。参考として例として、GPUのスケジューリングに関するKubernetesのドキュメントを参照してください。Sparkはリソース制限の設定のみをサポートしています。

Kubernetesは、各コンテナに割り当てられたリソースのアドレスをSparkに通知しません。そのため、ユーザーは、起動時にexecutorによって実行され、そのexecutorで使用可能なリソースを検出するディスカバリスクリプトを指定する必要があります。サンプルスクリプトは、examples/src/main/scripts/getGpusResources.shにあります。このスクリプトには実行権限を設定する必要があり、ユーザーは悪意のあるユーザーがスクリプトを変更できないようにパーミッションを設定する必要があります。スクリプトは、ResourceInformationクラスの形式でJSON文字列をSTDOUTに出力する必要があります。これには、リソース名と、そのexecutorのみが使用できるリソースアドレスの配列が含まれます。

リソースレベルスケジューリングの概要

Kubernetes上のSparkでは、いくつかのリソースレベルのスケジューリング機能がサポートされています。

優先順位スケジューリング

Kubernetesは、デフォルトでPod優先度をサポートしています。

Kubernetes上のSparkでは、Podテンプレートによってジョブの優先度を定義できます。ユーザーは、driverまたはexecutorのPodテンプレートのspecセクションでpriorityClassNameを指定できます。指定方法の例を以下に示します。

apiVersion: v1
Kind: Pod
metadata:
  labels:
    template-label-key: driver-template-label-value
spec:
  # Specify the priority in here
  priorityClassName: system-node-critical
  containers:
  - name: test-driver-container
    image: will-be-overwritten

Kubernetes 上の Spark 用のカスタマイズされた Kubernetes スケジューラ

Sparkでは、ユーザーはカスタムKubernetesスケジューラを指定できます。

1. スケジューラ名を指定します。

   ユーザーは、spark.kubernetes.scheduler.nameまたはspark.kubernetes.{driver/executor}.scheduler.name設定を使用してカスタムスケジューラを指定できます。

2. スケジューラ関連の設定を指定します。

   カスタムスケジューラを設定するには、Podテンプレートを使用し、ラベル（spark.kubernetes.{driver,executor}.label.*）、アノテーション（spark.kubernetes.{driver/executor}.annotation.*）、またはスケジューラ固有の設定（spark.kubernetes.scheduler.volcano.podGroupTemplateFileなど）を追加できます。

3. スケジューラ機能ステップを指定します。

   ユーザーは、spark.kubernetes.{driver/executor}.pod.featureStepsを使用して、以下を含むより複雑な要件をサポートすることもできます。

   • driver/executorスケジューリングのために追加のKubernetesカスタムリソースを作成します。
   • 設定または既存のPod情報に基づいて、スケジューラヒントを動的に設定します。

Kubernetes 上の Spark 用のカスタマイズされたスケジューラとしての Volcano の使用

前提条件

• カスタムスケジューラとしてVolcanoを使用したKubernetes上のSparkは、Spark v3.3.0およびVolcano v1.7.0以降でサポートされています。Volcano 1.7.0のインストール例を以下に示します。

  kubectl apply -f https://raw.githubusercontent.com/volcano-sh/volcano/v1.7.0/installer/volcano-development.yaml
  

ビルド

Sparkのダウンロードページで配布されているような、Volcanoサポートを含むSparkディストリビューションを作成するには、“Building Spark”も参照してください。

./dev/make-distribution.sh --name custom-spark --pip --r --tgz -Psparkr -Phive -Phive-thriftserver -Pkubernetes -Pvolcano

使用方法

Kubernetes上のSparkでは、カスタムスケジューラとしてVolcanoを使用できます。ユーザーはVolcanoを使用して、より高度なリソーススケジューリング（キュースケジューリング、リソース予約、優先順位スケジューリングなど）をサポートできます。

カスタムスケジューラとしてVolcanoを使用するには、次の設定オプションを指定する必要があります。

# Specify volcano scheduler and PodGroup template
--conf spark.kubernetes.scheduler.name=volcano
--conf spark.kubernetes.scheduler.volcano.podGroupTemplateFile=/path/to/podgroup-template.yaml
# Specify driver/executor VolcanoFeatureStep
--conf spark.kubernetes.driver.pod.featureSteps=org.apache.spark.deploy.k8s.features.VolcanoFeatureStep
--conf spark.kubernetes.executor.pod.featureSteps=org.apache.spark.deploy.k8s.features.VolcanoFeatureStep

Volcano 機能ステップ

Volcano機能ステップは、Volcano PodGroupを作成し、このPodGroupにリンクするdriver/executor podアノテーションを設定するのに役立ちます。

現在、Volcano機能ステップでは、driver/jobレベルのPodGroupのみがサポートされていることに注意してください。

Volcano PodGroup テンプレート

Volcanoは、CRD yamlを使用してPodGroup specを定義します。

Podテンプレートと同様に、SparkユーザーはVolcano PodGroupテンプレートを使用してPodGroup spec設定を定義できます。そのためには、Sparkプロパティspark.kubernetes.scheduler.volcano.podGroupTemplateFileを指定して、spark-submitプロセスからアクセス可能なファイルを指し示します。PodGroupテンプレートの例を以下に示します。

apiVersion: scheduling.volcano.sh/v1beta1
kind: PodGroup
spec:
  # Specify minMember to 1 to make a driver pod
  minMember: 1
  # Specify minResources to support resource reservation (the driver pod resource and executors pod resource should be considered)
  # It is useful for ensource the available resources meet the minimum requirements of the Spark job and avoiding the
  # situation where drivers are scheduled, and then they are unable to schedule sufficient executors to progress.
  minResources:
    cpu: "2"
    memory: "3Gi"
  # Specify the priority, help users to specify job priority in the queue during scheduling.
  priorityClassName: system-node-critical
  # Specify the queue, indicates the resource queue which the job should be submitted to
  queue: default

Kubernetes 上の Spark 用のカスタマイズされたスケジューラとしての Apache YuniKorn の使用

Apache YuniKornは、ジョブキューイング、リソース公平性、最小/最大キュー容量、柔軟なジョブ順序付けポリシーなどの高度なバッチスケジューリング機能を提供するKubernetesのリソーススケジューラです。利用可能なApache YuniKorn機能については、コア機能を参照してください。

前提条件

Apache YuniKornをインストールします。

helm repo add yunikorn https://apache.github.io/yunikorn-release
helm repo update
helm install yunikorn yunikorn/yunikorn --namespace yunikorn --version 1.3.0 --create-namespace --set embedAdmissionController=false

上記の手順により、既存のKubernetesクラスタにYuniKorn v1.3.0がインストールされます。

はじめに

次の追加オプションを使用してSparkジョブを送信します。

--conf spark.kubernetes.scheduler.name=yunikorn
--conf spark.kubernetes.driver.label.queue=root.default
--conf spark.kubernetes.executor.label.queue=root.default
--conf spark.kubernetes.driver.annotation.yunikorn.apache.org/app-id={{APP_ID}}
--conf spark.kubernetes.executor.annotation.yunikorn.apache.org/app-id={{APP_ID}}

{{APP_ID}}は、SparkジョブIDで自動的に置き換えられる組み込み変数であることに注意してください。上記の設定により、ジョブはデフォルトのKubernetesスケジューラではなく、YuniKornスケジューラによってスケジュールされます。

ステージレベルスケジューリングの概要

ステージレベルのスケジューリングはKubernetesでサポートされています。

• 動的割り当てが無効になっている場合：ユーザーはステージレベルで異なるタスクリソース要件を指定でき、起動時に要求されたのと同じexecutorを使用します。
• 動的割り当てが有効になっている場合：ユーザーはステージレベルでタスクとexecutorのリソース要件を指定でき、追加のexecutorを要求します。これには、Kubernetesは現時点では外部シャッフルサービスをサポートしていないため、spark.dynamicAllocation.shuffleTracking.enabledを有効にする必要もあります。異なるプロファイルのコンテナがKubernetesから要求される順序は保証されません。Kubernetesでの動的割り当てにはシャッフル追跡機能が必要であるため、異なるResourceProfileを使用した以前のステージからのexecutorは、シャッフルデータがあるためにアイドルタイムアウトにならない可能性があります。これにより、クラスタリソースの消費量が増え、最悪の場合、Kubernetesクラスタにリソースが残っていないと、Sparkがハングする可能性があります。spark.dynamicAllocation.shuffleTracking.timeout設定を使用してタイムアウトを設定することを検討できますが、シャッフルデータが本当に必要な場合は、データの再計算が必要になる可能性があります。基本のデフォルトプロファイルとカスタムResourceProfileの間では、podテンプレートリソースの処理方法に違いがあることに注意してください。podテンプレートファイルに指定されているリソースは、基本のデフォルトプロファイルでのみ使用されます。カスタムResourceProfileを作成する場合は、テンプレートファイルからのリソースはカスタムResourceProfileに伝播されないため、必要なリソースをすべて含めるようにしてください。