The optimization practice of Spark on k8s in Alibaba Cloud EMR

K8s 上の Spark の基本概念と特徴

まず、Spark on K8s の背景についてご紹介します。

1. Spark のクラスターデプロイモード

Spark は現在、4 つのデプロイモードをサポートしています。

・Standalone：Spark 組み込みのスケジューラを使用します。主にテスト環境で利用されますが、ビッグデータのスケジューリングフレームワークの機能を十分に活用できないため、クラスターリソースを十分に活用できません。

・Hadoop YARN：最も一般的な方法で、Hadoop に由来し、コミュニティのエコシステムが充実しています。

・Apache Mesos：YARN と同様のリソース管理フレームワークですが、徐々に歴史の表舞台から退いています。

・Kubernetes：Spark on K8s です。Spark 3.1.1 で公式に利用可能なサポートが提供され、ますます多くのユーザーが積極的にこのデプロイモードを試しています。

Spark on K8s を使用する利点は以下の通りです。

・リソース利用率の向上：利用シーンごとに複数のクラスターをデプロイする必要がなく、すべての Spark ジョブがクラスターリソースを共有することで、クラスター全体の利用率を向上できます。また、クラウド上で使用する場合、コンテナインスタンスを弾性的に伸縮でき、従量課金制で利用できます。

・運用保守方式の統一：Kubernetes のコミュニティエコシステムとツールを活用してクラスターを一元的に保守し、クラスター切り替えに伴う運用保守コストを削減できます。

・コンテナ化：コンテナイメージ管理により、Spark タスクの移植性を向上し、異なるバージョンの Spark によるバージョン競合を回避するとともに、複数バージョンでの A/B テストをサポートします。

特に、テスト結果によると、同一のクラスターリソース条件下では、Spark on K8s と Spark on YARN の性能差は無視できる程度です。さらに、Spark on K8s の弾性リソースを十分に活用することで、Spark ジョブをより効果的に高速化できます。

結論として、Spark on YARN モデルと比較して、Spark on K8s モデルの長所は短所を上回ります。

2. Spark on K8s のデプロイアーキテクチャ

現在の環境では、Spark ジョブを K8s に送信する方法が 2 つあります。

・ネイティブの spark-submit を使用する方法

この方法では、K8s クラスターに事前にコンポーネントをインストールする必要がありません。現在の YARN への送信方式と同様に、ジョブを送信するクライアント側に Spark 環境をインストールし、K8s クラスター接続ツールである kubectl を設定する必要があります。そして、送信コマンドで K8s クラスターのアドレスと使用する Spark イメージのアドレスを指定します。

上記の図は、ネイティブの spark-submit を使用して K8s にタスクを送信する際のタスク実行プロセスを詳細に示しています。クライアント側で spark-submit コマンドを実行すると、ローカルでプロセスが起動されます。このプロセスは K8s の API サーバーに接続し、Driver Pod の作成をリクエストします。Driver Pod は起動プロセスで Spark Context を開始し、Executor Pod の申請を担当します。タスク完了後、Driver Pod が Executor Pod のクリーンアップを担当します。ただし、Driver Pod は完了後もログやステータスの確認用に保持されるため、手動でのクリーンアップが必要です。

利点：

この送信方法はユーザーの利用習慣に合致し、学習コストを削減でき、既存のビッグデータプラットフォームとの統合も容易です。Client モードでの送信のため、ローカル依存関係および Spark シェルのインタラクティブ操作モードをサポートしています。

・Spark-on-K8s operator を使用する方法

Spark-on-K8s operator は Google のオープンソースコンポーネントです。事前に K8s クラスター内に常駐 Pod をデプロイして関連サービスを提供する必要があります。最初の方法とは異なり、コマンドラインでの送信は行わず、kubectl を使用して YAML ファイルを送信し、ジョブを実行します。本質的には、このツールは spark-submit の形式で実装されており、コマンドラインの情報をファイル形式で送信することに相当します。ただし、Spark-on-K8s operator は最初の方法をベースに、スケジュール実行、モニタリング、ジョブ管理などの補助ツールを追加しています。

プロセスとしては、ユーザーが YAML ファイルを送信すると、K8s クラスターに常駐する Spark on K8s operator がイベントをリッスンし、ファイルを解析して spark-submit コマンドを実行することで Spark タスクを起動します。

送信方法の違いに加え、前述の通りこのツールはいくつかの補助機能を提供します。Spark on K8s operator は K8s の Mutating Admission Webhook メカニズムを通じて K8s の API リクエストをインターセプトします。Driver Pod と Executor Pod のリソース起動時にカスタマイズが可能です。また、Driver Pod と Executor Pod のイベントをモニタリングして、タスク実行の進捗を追跡・管理できます。

利点：

ツールの存在により、記録、リトライ、スケジュール実行を含むジョブ管理がサポートされます。ジョブモニタリング指標を提供し、Prometheus と連携して一元的なモニタリングも可能です。ジョブリソースの自動クリーンアップをサポートし、Spark UI の Service/Ingress の自動設定も可能です。

3. Spark on K8s のコミュニティの進捗

Spark 2.3 以前にも、K8s 上に YARN をデプロイすることで Spark on K8s をサポートする試みがありましたが、本質的には Spark は依然として YARN のリソース管理下にあったため、完全な意味での Spark on K8s とは言えませんでした。

Spark 2.3 は、コミュニティがネイティブの Spark on K8s サポートを初めてリリースしたバージョンであり、このデプロイモードを正式にサポートした最初の事例です。

Spark 2.4 はいくつかの機能最適化を行い、Spark 3 で機能が大幅に改善されました。特に Spark 3.1 で正式に GA（一般提供）となった時点で大きく進歩しました。現在、Spark on K8s の方向性は非常に人気が高いため、興味のあるユーザーは Spark 3.1 に直接アップグレードしてこのデプロイ方法を試すことをお勧めします。

4. Spark on K8s の主要機能

・Spark Pod の設定プロパティの最適化

K8s の Pod 定義は通常 YAML 記述を使用します。初期の Driver Pod と Executor Pod の定義は Spark Conf 経由でのみ設定可能でしたが、すべての設定を Spark Conf で処理できるわけではありません。Spark 3.0 からはテンプレートファイルがサポートされ、ユーザーがテンプレートファイルを作成して Pod の属性を定義し、Spark の設定経由で渡すことができます。単一の設定と比較して、より便利で柔軟です。

・動的リソース割り当て

Spark 2 では、動的リソース割り当てに External Shuffle Service (ESS) モードのみ使用できました。このモードでは、エグゼキューターが実行中に生成したすべてのシャッフルデータを ESS サービスが引き受け、エグゼキューターは実行後いつでもリサイクルできます。ただし、このモードは通常 YARN の Node Manager によって起動・管理されるため、K8s 上でのデプロイが困難です。

Spark 3 は Shuffle Tracking 機能をサポートし、ESS なしでエグゼキューター自身の管理により動的リソース割り当ての効果を実現できます。ただし、この方法の欠点は、シャッフルリードフェーズ中にエグゼキューターを動的にリサイクルできないことです。Reducer がシャッフルデータを読み取るためにエグゼキューターを保持しておく必要があり、Driver 側の GC がエグゼキューター解放可能とマークするまで待機する必要があります。リソース解放の効率は低くなります。

・ノードデコミッショニング

K8s 環境では、ノードの縮小やプリエンプティブインスタンスの回収は一般的な現象です。特に、一部の Spark タスクの優先度を下げて他の高優先度タスクの利用に対応するシナリオでは、エグゼキューターが直接終了する際にステージの再計算が発生し、Spark の実行時間が延びる可能性があります。Spark 3.1 は「グレースフルシャットダウン」機能を提供します。Executor Pod が「強制」オフラインになる前に、Driver に新しいタスクを割り当てないよう通知し、キャッシュデータやシャッフルファイルを他の Executor Pod にマイグレーションして、対応する Spark タスクの効率を確保し、再計算を回避できます。

現在、この機能は実験的であり、デフォルトでは有効化されていません。

・PersistentVolumeClaim の再利用

Persistent Volume Claim（略称 PVC）は K8s のストレージ宣言です。各 Pod は明示的に申請してマウントできます。Spark 3.1 は PVC の動的作成をサポートし、事前に宣言する必要がなく、アプリケーションの実行に伴ってリソースを動的にアタッチできます。ただし、この時点での PVC のライフサイクルは Executor に伴います。前述のプリエンプティブな強制終了が行われた場合、PVC に保存されたデータも失われ、再計算が必要になります。そのため、Spark 3.2 は PVC の再利用をサポートしました。ライフサイクルが Driver に伴うようになり、再申請と再計算を回避し、全体の効率を確保します。

Alibaba Cloud EMR における Spark on K8s の最適化とベストプラクティス

次に、Alibaba Cloud EMR の Spark on K8s に対する最適化とベストプラクティスについてご紹介します。

1. Spark on ACK の概要

ACK：Alibaba Cloud の Container Service for Kubernetes、略称 ACK です。

EMR：Alibaba Cloud のオープンソースビッグデータプラットフォーム E-MapReduce、略称 EMR です。

Alibaba Cloud のパブリッククラウドには、Spark タイプのクラスターを含む EMR on ACK プロダクトがあります。以下、Spark on ACK と呼びます。Spark on ACK はセミマネージド型のビッグデータプラットフォームです。まず、ユーザーは自身の ACK クラスター（K8s クラスター）を持つ必要があります。そして、このクラスター内に Spark ジョブ用の名前空間を作成し、Spark Operator や HistoryServer などの固定コンポーネントをインストールします。以降の Spark ジョブ Pod もこの名前空間の下で実行されます。これらの Spark ジョブ Pod は、ユーザー自身の ACK ノードマシンを使用して実行することも、弾性インスタンス ECI を使用して従量課金制で実行することもできます。弾性インスタンス ECI については、詳しくご紹介します。

2. クラウド上の弾性の優位性

Spark on クラウドの最大の利点は、より優れた弾性です。Alibaba Cloud の ACK 環境では、弾性コンテナインスタンス ECI プロダクトが提供されています。ECI とは、Pod を申請する際に自身のマシンノードのリソースを使用するのではなく、クラウド上のリソースを完全に使用して Pod を作成することを意味します。素早く起動でき、秒単位での課金が可能です。ECI を使用して Spark ジョブを実行するのは非常にコスト効率が良いと考えます。通常、Spark ジョブはバッチ処理タスクの実行に使用され、早朝のピーク時には大量の処理が行われますが、日中は数件のクエリしかない場合があります。この明らかなピークと谷の特徴は、高速な弾性と従量課金制に非常に適しています。さらに、ECI はスポットプリエンプティブインスタンスを使用でき、1 時間の保護期間があり、Spark のノードデコミッショニング機能と組み合わせることで、大幅なコスト削減が可能です。

3. RSS によるシャッフルと動的リソースの最適化

Spark Shuffle はローカルストレージに大きく依存します。しかし、クラウド環境では、専用ストレージを持つマシンが自身のローカルディスクを保証することが困難です。クラウドディスクのサイズは予測が難しく、コストパフォーマンスも高くありません。一方で、Spark の ESS なしのネイティブな動的リソース割り当てでは、エグゼキューターのリソース解放効率が低く、リサイクルできないためリソースの浪費を招く可能性があります。

Spark Shuffle 自体にも多くの欠点があります。Mapper の出力が増加し、ローカルディスクへのスパイクを引き起こして追加の IO を発生させます。Reducer が Mapper 側のデータを並列にプルするため、大量のランダムリードが発生し、効率が低下します。シャッフルプロセスで numMapper × numReducer のネットワーク接続が生成され、CPU リソースを過剰に消費してパフォーマンスと安定性の問題を引き起こします。シャッフルデータの複製時にデータ損失が発生すると、再計算が必要となり、リソースを浪費します。

Alibaba Cloud は独立デプロイ型の RSS を提供しています。現在 GitHub でオープンソース化されており、ACK に直接接続できます。ユーザーはシャッフルデータがローカルディスクでサポートされているかどうかを気にする必要がありません。従来の Spark Shuffle ではシャッフルデータがエグゼキューターのローカルディスクに保存されていましたが、RSS 使用後はシャッフルデータが RSS に管理を委任されます。実際、プッシュベースの外部シャッフルサービスの使用については業界でコンセンサスが形成されており、多くの企業がこの分野で最適化に取り組んでいます。多くの利点があります。エグゼキューターは実行後にリサイクルしてリソースを節約できます。RSS は従来の大量のランダムリードを追加の書き込みと順次リードに最適化し、Spark Shuffle の効率問題をさらに補います。RSS サービスは HA デプロイとマルチコピーモードをサポートし、再計算の可能性を低減し、Spark タスクの効率をさらに確保します。

4. K8s のジョブレベルスケジューリングの強化

K8s スケジューラのデフォルトのスケジューリング粒度は Pod ですが、従来の Spark タスクのデフォルトのスケジューリング粒度はアプリケーションです。1 つのアプリケーションの起動には複数の Pod の起動が伴います。そのため、大量の Spark タスクが一度に送信されると、多数の Driver Pod が起動し、すべてが Executor Pod の起動を待機する状態になり、クラスター全体のデッドロックを引き起こす可能性があります。一方で、K8s のマルチテナントシナリオは十分にサポートされておらず、テナント間の柔軟なスケジューリングや動的クォータもサポートされていません。YARN のスケジューリング戦略と比較して、K8s のスケジューリング戦略は単一で、デフォルトの優先度 + FIFO モードであり、公平スケジューリングを実現できません。

Alibaba Cloud ACK はこの点で強化されています。

・スケジューリング時にリソースが十分かどうかを優先的に判断し、前述のデッドロック問題を解決します。

・名前空間ベースのマルチテナントツリーキューを実装しています。キューに対してリソースの上限と下限を設定でき、キュー間のリソースプリエンプションをサポートします。

・Spark ジョブの優先度キューはアプリケーション粒度でスケジューリングされ、キュー間の公平スケジューリングをサポートします。Spark-on-K8s operator の拡張に基づき、送信されたジョブは自動的にキューに入ります。

5. クラウドデータレイクストレージとアクセラレーション

・K8s 環境では、従来の Hadoop クラスターと比較して、データレイクストレージ OSS の方がストレージとコンピュートの分離アーキテクチャに適しています。Spark on ACK には Jindo SDK が組み込まれており、OSS とシームレスに接続できます。

・Fluid は Spark on K8s デプロイモードでのキャッシュアクセラレーションをサポートし、TPC-DS シナリオで実行速度を約 30% 向上できます。

6. DLF を使用したクラウドデータレイクの構築

K8s 上で Hadoop エコシステムのコンポーネントを使用する場合、追加のデプロイが必要です。しかし、Spark on ACK は Alibaba Cloud DLF（Data Lake Formation）とシームレスに接続されています。DLF は統合メタデータサービスを提供し、権限制御と監査をサポートするほか、データウェアハウジング、Spark SQL のインタラクティブ分析、データレイク管理、ストレージ分析、コスト最適化も提供します。

7. 使いやすさの向上

Spark on ACK は CLI ツールを提供しており、spark-submit の構文で直接 Spark ジョブを送信でき、同時に Spark Operator にも記録されて管理されます。前述の 2 つのジョブ送信方法の長所と短所について触れましたが、Spark Operator は優れたジョブ管理機能を持つ一方で、送信されるジョブは従来のコマンド構文と互換性がなく、インタラクティブシェルも実行できません。古いクラスターから移行するユーザーは変更に手間がかかります。そこで、このツールを使用することで、両方の方法の利点を同時に享受でき、ユーザーの利便性を大きく向上させます。

ログ収集については、Spark on ACK はログ収集ソリューションを提供し、HistoryServer を通じて Spark on YARN と同様にインターフェース上でログを確認できます。