データの溢れ - なぜ起きるのか、どのように排除するのか

Spark SQLのシャッフルパーティション(join、集計などのワイドの変換処理を実行するために使用されるCPUの数)の数のデフォルト設定は200であり、これが常にベストな値とは限りません。この結果、処理する大量のデータに対してそれぞれのSparkタスク(CPUコア)が割り当てられ、それぞれのコアがそのデータにフィットするのに十分なメモリーがない場合、それらの幾らかはディスクに溢れます。溢れ(スピル)はいかなる労力を払っても回避すべきであり、そのためにはシャッフルパーティションの数をチューニングしなくてはなりません。以下で議論するように、Spark SQLのシャッフルパーティションの数をチューニングするにはいくつかの方法があります。

1. AQEのオートチューニング

Spark AQEには、シャッフルパーティションの適切な値を自動で特定できるautoOptimizeShuffle (AOS)という機能があります。オートチューニングを有効化するための設定は以下の通りとなります。

set spark.sql.shuffle.partitions=auto

注意異常な圧縮

AOSには特定の制限が存在します。AOSはソーステーブルで通常ではないような圧縮率(20倍から40倍)がある場合において、適切なシャッフルパーティションを推定することができない場合があります。

圧縮率の高いテーブルを特定するには2つの方法があります:

1. Spark UIのSQL DAG

Exchangeノードのdata size totalメトリクスはメモリーにおけるテーブルの正確なサイズを示しませんが、圧縮率の高いテーブルを特定する助けになります。Scan Parquetノードは、ディスクにおけるテーブルの正確なサイズを提供します。上述のケースにおいて、Exchangeノードのデータサイズはディスクのサイズよりも40倍大きくなっており、テーブルはおそらくディスクでの圧縮率が高くなっていることを示しています。

2. テーブルのキャッシュ

メモリーにおける実際のサイズを特定するために、テーブルをメモリーにキャッシュすることができます。以下に方法を示します:

-- count here is forcing the cache materialization
spark.table("table").cache().count()

上のコマンドを実行した後に、メモリーにおけるテーブルのサイズを特定するには、Spark UIのストレージタブを参照します。

ソリューション: この現象に対応するには、以下のように初期のシャッフルパーティションの値(デフォルトは128MB)を特定するために、AQEによって使用されるパーティションあたりの値を少なくします:

-- setting to 16MB for example
set spark.databricks.adaptive.autoOptimizeShuffle.preshufflePartitionSizeInBytes = 16777216

preshufflePartitionSizeInBytesの値を16MBを小さくすると、AOSがパーティションの値を間違って計算していており、大きなデータの溢れを経験しているのであれば、preshufflePartitionSizeInBytesをさらに8MBに削減します。これでも溢れの問題を解消できない場合には、AOSを無効化して、次のセクションで説明するようにシャッフルパーティションの数を手動で調整するのがベストと言えます。

2. 手動でのファインチューン

シャッフルパーティションの数を手動でチューニングするには、以下を必要とします:

シャッフルされるデータの総量。そのためには、一度Sparkクエリーを実行し、以下の例に示しているようにこの値を取得するためにSpark UIを使います。
経験則として、シャッフルパーティションの数を調整した後には、それぞれのタスクがおおよそ128MBから200MBのデータを処理するようになっていることを確認する必要があります。以下の例で示しているように、Spark UIのシャッフルステージのサマリーメトリクスでこの値を確認することができます。

こちらが、シャッフルパーティションの適切な値を計算するための数式となります:

以下を仮定しましょう:

クラスターにおける総ワーカーコアの合計数 = T
シャッフルステージでシャッフルされるデータの総量(MB) = B
タスクごとに処理される最適なデータサイズ(MB) = 128
結果として、積算のファクター(M)は: M = ceiling(B / 128 / T)となり、
シャッフルパーティションの数(N)は: N = M x T になります。

最後の実行サイクルまで全てのクラスターコアが完全にエンゲージするように、ここではceiling関数を使っていることに注意してください。

タスクごとに処理される最適なデータのサイズは約128MBです。多くの場合、これでうまくいきます。あなたのクエリーである種のデータ爆発が発生している場合はうまくいかないかもしれません。その場合は、より小さい値を選択しなくてはならないかもしれません。本書の後のセクションでデータ爆発に関して取り扱っています。

シャッフルパーティションに対するオートチューン(AOS)も手動でのファインチューニングも行わない場合、経験則としてワーカーのCPUコア数の2倍、あるいは3倍にするというものがあります。

-- in SQL
set spark.sql.shuffle.partitions = 2*<number of total worker cores in cluster>
-- in PySpark
spark.conf.set(“spark.sql.shuffle.partitions”, 2*<number of total worker cores in cluster>)
-- or
spark.conf.set(“spark.sql.shuffle.partitions”, 2*sc.defaultParallelism)

単一のノートブックに複数のSpark SQLクエリーが存在する場合があるため、それぞれのクエリーに対するシャッフルパーティション数のファインチューニングは時間を浪費するタスクとなります。このため、シャッフルステージでシャッフルされるデータの総量が最も多い最大のクエリーに対してファインチューニングし、ノートブック全体にその値を適用するのがアドバイスとなります。
データに偏りがある場合には、シャッフルパーティションのファインチューニングがデータの溢れを改善しません。この場合、データの偏りを最初に取り除くべきです。詳細に関しては、次のデータ偏りに関するセクションをご覧ください。

はじめてのDatabricks

はじめてのDatabricks

Databricks無料トライアル

Databricks無料トライアル