Kubernetes で Spark を実行

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

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

セキュリティ

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

ユーザー ID

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

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

ボリュームマウント

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

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

前提条件

• バージョン >= 1.30 の実行中の Kubernetes クラスターと、kubectl を使用したアクセス設定。まだ動作中の Kubernetes クラスターがない場合は、minikube を使用してローカルマシンにテストクラスターを設定できます。
  • DNS アドオンが有効になっている minikube の最新リリースを使用することをお勧めします。
  • デフォルトの minikube 設定では、Spark アプリケーションを実行するには十分ではないことに注意してください。単一のエグゼキューターで単純な Spark アプリケーションを開始するには、3 CPU と 4GB のメモリをお勧めします。
  • Spark 環境の kubernetes-client library のバージョンと、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 サーバーは api_server_url で連絡されます。URL に HTTP プロトコルが指定されていない場合、デフォルトは https になります。たとえば、マスターを k8s://example.com:443 に設定することは、k8s://https://example.com:443 に設定することと同等ですが、TLS なしで別のポートに接続するには、マスターを k8s://http://example.com:8080 に設定する必要があります。

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

Kubernetes クラスターがセットアップされている場合、API サーバー URL を見つける 1 つの方法は、kubectl cluster-info を実行することです。

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

上記の例では、特定の Kubernetes クラスターを spark-submit で使用するために、--master k8s://http://127.0.0.1:6443 を spark-submit の引数として指定できます。さらに、認証プロキシである 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 アプリケーションを実行できるようになりました。アプリケーションがクライアントモードで実行される場合、ドライバーはポッド内または物理ホストで実行できます。クライアントモードでアプリケーションを実行する場合は、次の要因を考慮することをお勧めします。

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

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

クライアントモードの実行ポッドのガベージコレクション

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

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

spark.kubernetes.executor.podNamePrefix を使用して、エグゼキューターポッド名を完全に制御できます。このプロパティが設定されている場合、同じ名前空間内のすべてのジョブで一意にすることが強く推奨されます。

認証パラメータ

クライアントモードの Kubernetes 認証パラメータには、正確なプレフィックス spark.kubernetes.authenticate を使用します。

IPv4 および IPv6

3.4.0 以降、Spark は、Pod および Service に IPv4 および IPv6 アドレスを割り当てることができる IPv4/IPv6 デュアルスタックネットワーク機能を介して、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 のみを SingleStack で使用するには、次のようにジョブを送信できます。

...
    --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.4.1
--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 にアップロードされ、ドライバーが起動するときにドライバーポッドにダウンロードされてクラスパスに追加されます。Spark は、並行して実行される Spark アプリケーションとの競合を回避するために、アップロードパスの下にランダムな名前のサブディレクトリを生成します。ユーザーは必要に応じて作成されたサブディレクトリを管理できます。

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

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

シークレット管理

Kubernetes シークレットは、Spark アプリケーションが保護されたサービスにアクセスするための認証情報を提供するために使用できます。ユーザー指定のシークレットをドライバーコンテナにマウントするには、spark.kubernetes.driver.secrets.[SecretName]=<mount path> の形式の構成プロパティを使用できます。同様に、spark.kubernetes.executor.secrets.[SecretName]=<mount path> の形式の構成プロパティを使用して、ユーザー指定のシークレットをエグゼキューターコンテナにマウントできます。マウントされるシークレットは、ドライバーおよびエグゼキューターポッドと同じ名前空間にあると想定されていることに注意してください。たとえば、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 では、テンプレートファイルからポッドを定義できます。Spark ユーザーも同様にテンプレートファイルを使用して、Spark の構成でサポートされていないドライバーまたはエグゼキューターポッドの構成を定義できます。これを行うには、spark.kubernetes.driver.podTemplateFile および spark.kubernetes.executor.podTemplateFile の Spark プロパティを指定して、spark-submit プロセスからアクセス可能なファイルを示します。

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

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

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

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

Kubernetes ボリュームの使用

ユーザーは、ドライバーおよびエグゼキューターポッドに次の種類の Kubernetes ボリュームをマウントできます。

• hostPath: ホストノードのファイルシステムからポッドにファイルまたはディレクトリをマウントします。
• emptyDir: ポッドがノードに割り当てられたときに作成される、最初は空のボリューム。
• nfs: 既存の NFS (ネットワークファイルシステム) をポッドにマウントします。
• persistentVolumeClaim: PersistentVolume をポッドにマウントします。

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

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

--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 は、ポッド仕様の 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 中心のエグゼキューター割り当てを実行できるようになりました。これは、Spark がジョブが持つことができる作成済み PVC の総数をカウントし、ドライバーが最大数の PVC を所有している場合、新しいエグゼキューターの作成を保持することを意味します。これは、既存の PVC があるエグゼキューターから別 のエグゼキューターに移行するのに役立ちます。

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 ダッシュボードからもアクセスできます。

ログ収集システムが存在する場合、Spark ドライバーの Executors タブ UI で公開できます。たとえば、

spark.ui.custom.executor.log.url='https://log-server/log?appId=&execId='

この URL テンプレートにカスタム変数を追加できます。これは、次のような既存のエグゼキューター環境変数の値で入力されます。

spark.executorEnv.SPARK_EXECUTOR_ATTRIBUTE_YOUR_VAR='$(EXISTING_EXECUTOR_ENV_VAR)'
spark.ui.custom.executor.log.url='https://log-server/log?appId=&execId=&your_var='

ドライバー UI へのアクセス

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

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

次に、Spark ドライバー UI は https://:4040 でアクセスできます。

Apache Spark 4.0.0 以降、Driver UI は新しい構成を通じてドライバーログを表示する方法を提供します。

spark.driver.log.localDir=/tmp

次に、Spark ドライバー UI は https://:4040/logs/ でアクセスできます。オプションで、ログのレイアウトは次のように構成されます。

spark.driver.log.layout="%m%n%ex"

デバッグ

いくつかの種類の障害が発生する可能性があります。Spark 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 では、リソースクォータを使用して、個々の名前空間のリソース、オブジェクト数などに制限を設定できます。管理者 は、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

両方の操作はグロブパターンをサポートしています。たとえば、ユーザーは次を実行できます。

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

上記は、指定されたプレフィックスを持つすべてのアプリケーションをキルします。

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

今後の作業

現在開発中または開発が計画されている Spark on Kubernetes の機能がいくつかあります。これらの機能は、将来の Spark-Kubernetes 統合バージョンに最終的に取り込まれることが期待されます。

それらの一部は次のとおりです。

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

設定

Spark の構成については、構成ページを参照してください。以下の構成は Spark on Kubernetes に固有のものです。

Spark プロパティ

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

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

ポッドメタデータ

ポッド仕様

コンテナ仕様

以下は、ドライバーおよびエグゼキューターコンテナに影響します。ポッドスペック内の他のすべてのコンテナは影響を受けません。

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

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

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

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

Kubernetesは、各コンテナに割り当てられたリソースのアドレスをSparkに通知しません。そのため、ユーザーは、エグゼキューターが起動時に実行して、そのエグゼキューターが利用可能なリソースを発見するための発見スクリプトを指定する必要があります。例のスクリプトは`examples/src/main/scripts/getGpusResources.sh`で見つけることができます。スクリプトには実行権限を設定する必要があり、ユーザーは悪意のあるユーザーがそれを変更できないように権限を設定する必要があります。スクリプトは、ResourceInformationクラスの形式のJSON文字列をSTDOUTに書き込む必要があります。これには、リソース名と、そのエグゼキューターのみが利用可能なリソースアドレスの配列が含まれます。

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

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

優先度スケジューリング

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

Spark on Kubernetesでは、ポッドテンプレートによってジョブの優先度を定義できます。ユーザーは、ドライバーまたはエグゼキューターのポッドテンプレートの`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

Spark on Kubernetes 用のカスタム Kubernetes スケジューラー

Sparkは、ユーザーがカスタムKubernetesスケジューラを指定できるようにします。

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

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

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

   カスタムスケジューラを構成するために、ユーザーはポッドテンプレートを使用したり、ラベル（`spark.kubernetes.{driver,executor}.label.*`）、アノテーション（`spark.kubernetes.{driver/executor}.annotation.*`）、またはスケジューラ固有の設定（例：`spark.kubernetes.scheduler.volcano.podGroupTemplateFile`）を追加したりできます。

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

   ユーザーは、`spark.kubernetes.{driver/executor}.pod.featureSteps`を使用して、以下を含むがこれらに限定されない、より複雑な要件をサポートすることも検討できます。

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

Volcano を Spark on Kubernetes 用のカスタムスケジューラーとして使用する

前提条件

• Spark on KubernetesとVolcanoは、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ディストリビューションを作成するには、「Sparkのビルド」も参照してください。

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

使用法

Spark on Kubernetesは、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にリンクするのに役立ちます。

現在、Volcano Feature Stepではドライバー/ジョブレベルのPodGroupのみがサポートされていることに注意してください。

Volcano PodGroup テンプレート

Volcanoは、CRD yamlを使用してPodGroupスペックを定義します。

ポッドテンプレートと同様に、SparkユーザーはVolcano PodGroupテンプレートを使用してPodGroupスペック構成を定義できます。これを行うには、`spark-submit`プロセスからアクセス可能なファイルにポイントするようにSparkプロパティ`spark.kubernetes.scheduler.volcano.podGroupTemplateFile`を指定します。以下は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

Apache YuniKorn を Spark on Kubernetes 用のカスタムスケジューラーとして使用する

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.6.3 --create-namespace --set embedAdmissionController=false

上記の手順は、既存のKubernetesクラスタにYuniKorn v1.6.3をインストールします。

はじめに

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