Spark AI Summit 2020の注目点: Spark 3.0のパフォーマンス改善のイノベーション

原文(投稿日：2020/07/03)へのリンク

初めてオンラインで開催された先頃のSpark AI Summit 2020で、注目点は、Spark SQLの最適化やNVIDIA GPUアクセラレーションなど、Apache Spark 3.0のパフォーマンスを改善するためのイノベーションだった。

Spark 3.0はSpark SQLを最適化

冒頭の基調講演で、CTOのMatei Zaharia氏は、Spark API呼び出しの90％がSpark SQLエンジンを介して実行され、その結果、Apache SparkコミュニティはSparkのパッチの46％がSpark SQLの改善への投資だと説明した。Spark 3.0はSpark 2.4 (TPC-DSを使用) よりも約2倍高速だ。それはアダプティブクエリ実行 (適応性のあるクエリ実行)、動的パーティションプルーニング、およびその他の最適化によって実現される。

Spark SQLのアダプティブクエリ実行

Spark 2.2は、既存のルールベースのSQLオプティマイザにコストベースの最適化を追加した。Spark 3.0には、実行時アダプティブクエリ実行 (AQE) がある。AQEでは、クエリプランの完了したステージから取得したランタイム統計を使用して、残りのクエリステージの実行プランを再最適化する。Databricksベンチマークは、AQEを使用した場合に1.1倍から8倍の範囲のスピードアップをもたらした。

Spark 3.0 AQE最適化は次の機能を含む:

• シャッフルパーティションを動的に合体: Sparkでは、パーティションの数はタスクの数に対応する。最適な並列化を実現するには、パーティションのバランスを取り、大きすぎたり小さすぎたりしないようにする必要だ。パーティションはデータに依存し、ステージごとに異なる。AQEは、シャッフルファイルの統計を確認することで、実行時に隣接する小さなパーティションを大きなパーティションに結合し、クエリ集計のタスク数を減らすことができる。
• 結合戦略を動的に切替: AQEは、結合関係のサイズに基づいて、実行時に結合戦略を最適化できる。たとえば、AQEは、ソートマージ結合をブロードキャストハッシュ結合に変更できる。これは、結合の片側がメモリに収まるほど小さい場合にパフォーマンスが向上する。
• スキュー結合を動的に最適化: 大規模なデータセットを操作する場合の一般的な問題は、クエリのパフォーマンスを大幅に低下させる可能性のあるソートマージ結合でのデータスキューの処理だ。Spark 3.0で動的に最適化されたスキュー結合は、ランタイム統計を使用してパーティションサイズからスキューを検出し、スキューパーティションをより小さなサブパーティションに分割できる。

Spark SQLの動的パーティションプルーニング

Spark 2.x 静的述語プッシュダウンとパーティションプルーニングは、クエリ時にSparkが読み取るファイルとパーティションの数を制限するパフォーマンスの最適化だ。データをパーティション分割した後、特定のパーティションフィルター基準に一致するクエリは、Sparkがディレクトリとファイルのサブセットのみを読み取ることを許可することでパフォーマンスを向上させる。Spark 3.0の動的パーティションプルーニングを使用すると、Sparkエンジンは実行時に動的に推測できる。特定のクエリに対して読み取りおよび処理する必要があるテーブル内の特定のパーティションを、結合で別のテーブルをフィルタリングした結果のパーティション列値を特定することだ。

Spark 3.0GPUアクセラレーション

Apache Spark 3.xのGPUサポートの詳細セッションでは、Robert Evans氏とJason Lowe氏が、アクセラレータ対応スケジューリングの概要と、GPUで高速化されたSQL/DataFrame操作とSparkシャッフルをコード変更なしで有効にするApache SparkのRAPIDS Acceleratorについて説明した。

アクセラレータ対応スケジューリング

GPUはディープラーニングを加速するために広く使用されているが、データ処理には使用されていない。Sparkでのディープラーニングとデータ処理をより適切に統合するための主要なSparkイニシアチブであるProject Hydrogenの一部として、GPUはApache Spark 3.0でスケジュール可能なリソースになった。これにより、Sparkは指定された数のGPUでエグゼキュータをスケジュールでき、ユーザは各タスクに必要なGPUの数を指定できる。Sparkは、これらのリソースリクエストを、基盤となるクラスタマネージャ、Kubernetes、YARN、またはスタンドアロンに伝達する。ユーザは、クラスタマネージャによって割り当てられたGPUを検出できる検出スクリプトを構成することもできる。これにより、以前はユーザがSparkアプリケーションでのGPUスケジューリングの欠如を回避する必要があったが、GPUを必要とするMLアプリケーションの実行が大幅に簡素化される。

SQL/DataFrameの加速化

Spark 3.0は、行ではなく列バッチを使用してデータを処理するSQLオプティマイザプラグインをサポートしている。列データはGPUに対応しており、この機能は、SQLおよびDataFrameオペレータを高速化するためにRAPIDS Acceleratorがプラグインする機能だ。RAPIDS Acceleratorでは、Catalystクエリオプティマイザが変更され、RAPIDS APIで高速化できるクエリプラン内のオペレータ (主に1対1のマッピング) を識別し、クエリプラン実行時にSparkクラスタ内のGPUでそれらのオペレータをスケジュールするようになった。

シャッフルの加速化

データを値で並べ替え、グループ化、または結合するSpark操作は、ディスクI/O、データシリアル化、およびネットワークI/Oを含むシャッフルと呼ばれるプロセスで、ステージ間で既存のDataFrameから新しいDataFrameを作成するときに、パーティション間でデータを移動する必要がある。新しいRAPIDS Acceleratorシャッフルの実装では、UCXを活用してGPUデータ転送を最適化し、GPUにできるだけ多くのデータを保持し、CPUをバイパスしてGPUにGPUメモリのノード内およびノード間転送を実行することや、利用可能な最善のハードウェアリソースを使用して、ノード間でデータを移動する最速のパスを見つける。

MLとDLのエンド間の加速化

Travis Addair氏とThomas Graves氏は、Apache SparkのHorovodを使用したエンド間のディープラーニングのセッションで、単一のパイプラインでETLとDLについて説明した。TensorFlow、Keras、PyTorch、Apache MXNetの分散型ディープラーニングトレーニングフレームワークであるHorovodは、Spark 3.0とGPUスケジューリングのサポートを追加した。Horovodは、Sparkとの統合と使いやすさを向上させるためにSpark MLパイプラインでSpark Estimatorsを使用する新しいKerasEstimatorクラスも追加した。これにより、TensorFlowモデルとPyTorchモデルをSpark DataFrameで直接トレーニングできるようになり、分散トレーニング用の特別なコードなしで、Horovodの機能を活用して数百のGPUに並行して拡張できる。Apache Spark 3.0の新しいアクセラレータ対応のスケジューリングAPIと列処理APIを使用すると、プロダクションETLジョブは、同じパイプライン内のGPUで分散ディープラーニングトレーニングを実行しているHorovodにデータを渡すことができる。

Rong Ou氏とBobby Wang氏は、Apache Spark GPUクラスタでXGBoostトレーニングのスケーラブルアクセラレーションのセッションでXGBoostのイノベーションについて説明した。XGBoostは、スケーラブルな分散型勾配ブースト決定木 (GBDT) MLライブラリだ。XGBoostはSpark 3.0のRapids Acceleratorと統合され、次のエンド間のGPUアクセラレーションを実現する: Spark SQL/DataFrameオペレータとXGBoostトレーニング時間、データセットのスパース性に基づいてメモリ内に最適に保存された機能、および改善されたGPUメモリの利用。

詳細について

ビデオと多くのセッションスライドを含む完全なスケジュールは、Spark AI Summit 2020のページにある。Spark 3.0の新機能の詳細は、Spark 3.0リリースノートに記載されている。Apache SparkのRAPIDS Acceleratorにアクセスするには、nvidia/spark-rapids GitHubリポジトリがある。Databricksは最近、Databricks Runtime 7.0の一部としてのSpark 3.0.0を、機械学習のDatabricks Runtime 7.0の一部としてGPUアクセラレーションを事前構成したSpark 3.0.0を利用できることを発表した。これは、Microsoft AzureとAWSでも利用できる。Google Cloudは最近、Dataprocイメージバージョン2.0でSpark 3.0プレビューが利用可能になったことを発表した。