翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

シャッフルの最適化

join() や などの特定のオペレーションではgroupByKey()、Spark でシャッフルを実行する必要があります。シャッフルは、RDD パーティション間で異なるグループ化になるようにデータを再分散するための Spark のメカニズムです。シャッフルは、パフォーマンスのボトルネックの修復に役立ちます。ただし、シャッフルには通常 Spark エグゼキュター間でのデータのコピーが含まれるため、シャッフルは複雑でコストのかかる操作です。たとえば、シャッフルでは次のコストが発生します。

シャッフルは、Spark アプリケーションのパフォーマンス低下における最も重要な要因の 1 つです。中間データを保存すると、エグゼキュターのローカルディスクに領域が枯渇し、Spark ジョブが失敗する可能性があります。

CloudWatch メトリクスと Spark UI でシャッフルパフォーマンスを評価できます。

CloudWatch メトリクス

Shuffle Bytes 書き込み値が Shuffle Bytes Read と比較して高い場合、Spark ジョブは join()や などのシャッフルオペレーションを使用することがありますgroupByKey()。

Spark UI

Spark UI のステージタブで、シャッフル読み取りサイズ/レコードの値を確認できます。Executors タブでも確認できます。

次のスクリーンショットでは、各エグゼキュターは約 18.6 GB/4020000 レコードをシャッフルプロセスと交換し、合計シャッフル読み取りサイズは約 75 GB) になります。

シャッフルスピル (ディスク） 列には大量のデータスピルメモリがディスクに表示され、フルディスクやパフォーマンスの問題が発生する可能性があります。

これらの症状を観察し、パフォーマンス目標と比較してステージに時間がかかりすぎる場合、または Out Of Memoryまたは No space left on device エラーで失敗する場合は、次の解決策を検討してください。

結合の最適化

テーブルを結合する join()オペレーションは、最も一般的に使用されるシャッフルオペレーションですが、多くの場合、パフォーマンスのボトルネックです。結合はコストのかかる操作であるため、ビジネス要件に不可欠でない限り、使用しないことをお勧めします。次の質問をして、データパイプラインを効率的に使用していることを再確認します。

参加オペレーションがビジネス要件に不可欠であることを確認したら、要件を満たす方法で参加を最適化するための以下のオプションを参照してください。

参加前にプッシュダウンを使用する

結合を実行する前に、DataFrame 内の不要な行と列を除外します。これには次の利点があります。

# Default
df_joined  = df1.join(df2, ["product_id"])

# Use Pushdown
df1_select = df1.select("product_id","product_title","star_rating").filter(col("star_rating")>=4.0)
df2_select = df2.select("product_id","category_id")
df_joined  = df1_select.join(df2_select, ["product_id"])

DataFrame 結合を使用する

RDD API または DynamicFrame 結合の代わりに、SparkSQL、DataFrame、データセットなどの Spark の高レベル API を試してください。 SparkSQL DataFrame DynamicFrame DynamicFrame を DataFrame に変換するには、 などのメソッド呼び出しを使用しますdyf.toDF()。Apache Spark の「キートピック」セクションで説明したように、これらの結合オペレーションは、Catalyst オプティマイザによるクエリ最適化を内部的に活用します。

ハッシュ結合とヒントのシャッフルとブロードキャスト

Spark は、シャッフル結合とブロードキャストハッシュ結合の 2 種類の結合をサポートしています。ブロードキャストハッシュ結合はシャッフルを必要としないため、シャッフル結合よりも処理が少なくて済みます。ただし、小さなテーブルを大きなテーブルに結合する場合にのみ該当します。単一の Spark エグゼキュターのメモリに収まるテーブルを結合する場合は、ブロードキャストハッシュ結合の使用を検討してください。

次の図は、ブロードキャストハッシュ結合とシャッフル結合の大まかな構造とステップを示しています。

各結合の詳細は次のとおりです。

AWS Glue 3.0 以降では、アダプティブクエリ実行と追加のパラメータを有効にすることで、ブロードキャストハッシュ結合を自動的に活用できます。アダプティブクエリ実行は、いずれかの結合側のランタイム統計がアダプティブブロードキャストハッシュ結合しきい値よりも小さい場合に、ソートマージ結合をブロードキャストハッシュ結合に変換します。

AWS Glue 3.0 では、 を設定することで Adaptive Query Execution を有効にできますspark.sql.adaptive.enabled=true。アダプティブクエリの実行は、AWS Glue 4.0 でデフォルトで有効になっています。

シャッフルとブロードキャストハッシュ結合に関連する追加のパラメータを設定できます。

関連するパラメータの詳細については、「ソートマージ結合をブロードキャスト結合に変換する」を参照してください。

AWS Glue 3.0 以降では、シャッフルに他の結合ヒントを使用して動作を調整できます。

-- Join Hints for shuffle sort merge join
SELECT /*+ SHUFFLE_MERGE(t1) /  FROM t1 INNER JOIN t2 ON t1.key = t2.key;
SELECT /*+ MERGEJOIN(t2) /  FROM t1 INNER JOIN t2 ON t1.key = t2.key;
SELECT /*+ MERGE(t1) /  FROM t1 INNER JOIN t2 ON t1.key = t2.key;

-- Join Hints for shuffle hash join
SELECT /*+ SHUFFLE_HASH(t1) /  FROM t1 INNER JOIN t2 ON t1.key = t2.key;

-- Join Hints for shuffle-and-replicate nested loop join
SELECT /*+ SHUFFLE_REPLICATE_NL(t1) /  FROM t1 INNER JOIN t2 ON t1.key = t2.key;

バケット化を使用する

ソートマージ結合には、シャッフルとソート、マージの 2 つのフェーズが必要です。これらの 2 つのフェーズでは、一部のエグゼキュターがマージしていて、他のエグゼキュターが同時にソートされている場合、Spark エグゼキュターが過負荷になり、OOM とパフォーマンスの問題が発生する可能性があります。このような場合、バケット化を使用して効率的に参加できる場合があります。バケット化は、結合キーで入力を事前シャッフルおよび事前ソートし、ソートされたデータを中間テーブルに書き込みます。ソートされた中間テーブルを事前に定義することで、大きなテーブルを結合する場合のシャッフルステップとソートステップのコストを削減できます。

バケットテーブルは、次の場合に便利です。

バケットテーブルを作成するには、次のコードを使用します。

df.write.bucketBy(50, "account_id").sortBy("age").saveAsTable("bucketed_table")

結合前に結合キーで DataFrames を再パーティションする

結合の前に結合キーの 2 つの DataFramesを再パーティションするには、次のステートメントを使用します。

df1_repartitioned = df1.repartition(N,"join_key")
df2_repartitioned = df2.repartition(N,"join_key")
df_joined  = df1_repartitioned.join(df2_repartitioned,"product_id")

これにより、結合を開始する前に、結合キーで 2 つの (まだ分離RDDs がパーティション化されます。2 つの RDDs が同じパーティション分割コードを持つ同じキーでパーティション分割されている場合、RDD は、結合のシャッフル前に、結合する計画が同じワーカーにコロケーションされる可能性が高くなります。これにより、結合中のネットワークアクティビティとデータスキューを減らすことで、パフォーマンスを向上させることができます。

データスキューを克服する

データスキューは、Spark ジョブのボトルネックの最も一般的な原因の 1 つです。これは、データが RDD パーティション間で均一に分散されていない場合に発生します。これにより、そのパーティションのタスクが他のパーティションよりもはるかに時間がかかり、アプリケーションの全体的な処理時間が遅延します。

データスキューを特定するには、Spark UI で次のメトリクスを評価します。

データスキューが表示された場合は、以下を試してください。

キャッシュを使用する

反復的な DataFrames を使用する場合は、 df.cache() または を使用して計算結果を各 Spark エグゼキュターのメモリとディスクにdf.persist()キャッシュすることで、追加のシャッフルや計算を回避します。Spark は、ディスクへの RDDs保持または複数のノード (ストレージレベル) へのレプリケーションもサポートしています。

たとえば、 を追加して DataFrames を保持できますdf.persist()。キャッシュが不要になった場合は、 unpersistを使用してキャッシュされたデータを破棄できます。

df = spark.read.parquet("s3://<Bucket>/parquet/product_category=Books/")
df_high_rate = df.filter(col("star_rating")>=4.0)
df_high_rate.persist()

df_joined1 = df_high_rate.join(<Table1>, ["key"])
df_joined2 = df_high_rate.join(<Table2>, ["key"])
df_joined3 = df_high_rate.join(<Table3>, ["key"])
... 
df_high_rate.unpersist()

不要な Spark アクションを削除する

count、show、 などの不要なアクションを実行しないでくださいcollect。Apache Spark の「キートピック」セクションで説明したように、Spark は遅延しています。変換された各 RDD は、アクションを実行するたびに再計算される場合があります。多くの Spark アクションを使用すると、アクションごとに複数のソースアクセス、タスク計算、シャッフル実行が呼び出されます。

商用環境で collect()やその他のアクションが必要ない場合は、それらを削除することを検討してください。