出題の文脈と適用条件
日次のSpark SQLジョブが、4.8 TBのイベントファクトテーブルと180 GBの製品ディメンションをproduct_idでLeft Joinしています。 上流のリリース後、イベントの32%がproduct_id='UNKNOWN'に正規化されるようになりました。このジョブは2,000の シャッフルパーティションを使用しています。Joinステージにおいて、Spark UIではタスクのシャッフル読み取り量の中央値が1.1 GiBであるのに対し、1つの タスクが720 GiBを読み取り、ディスクへのスピルを繰り返し、リトライ後に最終的にOOMで失敗します。実行時間は 24分から96分に増加しました。ビジネス要件として、不明な製品イベントを破棄したりLeft Joinの結果を変更したりすることなく、 ジョブを45分以内に完了させる必要があります。
テーブルサイズ、キーの割合、パーティションメトリクス、実行時間、およびSLAは面接上の前提条件です。中心となる タスクは、パーティションレベルのエビデンスを用いてデータスキューをリソース不足やJoin出力の爆発と区別し、 データコントラクトを保持する解決策を選択することです。これはSparkの実行計画、シャッフルパーティション、 データ分布、およびバッチの正当性をテストするため、データカテゴリに属します。既存のKafkaホットパーティションの 問題はメッセージキー、順序付け、コンシューマーオフセットに焦点を当てています。本問はランタイムSQLパーティション、 Join戦略、AQE、結果の保存性に焦点を当てているため、障害レイヤーと検証方法が異なります。
同じ論理はgroupBy、distinct、ウィンドウ関数、その他のWide Dependency（シャッフルを伴う依存関係）にも適用されます。 1つのキーに対する多数のレコードがシャッフル後の少数のタスクに集中すると、Straggler（遅延タスク）、スピル、GCプレッシャー、 またはOOMが発生する可能性があります。優れた回答は、安易にメモリを追加することから始めるのではなく、 最も遅いタスクが不均衡な量のデータと計算を抱えているかどうかをまず立証します。
面接官が評価するポイント
エビデンスの粒度から始めます。優れた回答は、ジョブからSQLクエリ、特定のステージ、そして個々のタスクへと掘り下げていきます。 所要時間、シャッフル読み取りのレコード数とバイト数、スピル、ピーク実行メモリ、GC時間を比較します。中央値の数百倍を読み取り、 別のExecutorでリトライしても依然として遅いタスクは、決定論的なデータスキューを裏付けます。Executor全体のメモリ総量や 合計実行時間だけでは、その原因を特定できません。
実行計画の理解力も同様に重要です。EXPLAIN FORMATTEDはExchange、Joinタイプ、および物理Joinを確認します。 EXPLAIN COSTとSQL UIのランタイム統計は、推定および観測されたデータサイズを明らかにします。 Spark AQEはランタイム統計を使用して計画を調整します。Spark 4.2.0では、Skew Join最適化により、スキューしたSort-Merge-Joinパーティションを 分割し、必要に応じて小さい側を複製できます。ドキュメント化されたデフォルト値では、中央値の5倍かつ256 MiBを 超えた場合にのみパーティションがスキューしていると判定されます。ドキュメントのデフォルト値はクラスタの不変の事実ではないため、 候補者は有効な環境設定を調査する必要があります。
データのセマンティクスが重要な分岐点となります。UNKNOWNは、有効な「属性未確定」イベントである場合もあれば、上流の不具合である場合もあります。 それらの行を破棄、ランダム分散、または書き換えると、結果が変わる可能性があります。ソルティング（Salting）を用いたJoinでは、 ディメンション側のホットキーの行のみを複製し、ファクト側のホットな行に対して決定論的なソルトを割り当てる必要があります。ディメンション全体を 複製するとデータ量が何倍にも増加し、一方で両側を独立してランダム化すると一致が失われます。
救済策の選択によって、さらに深い知識が明らかになります。spark.sql.shuffle.partitionsを増やすとハッシュバケットは増えますが、 1つのホットキーに対するすべての行は依然として1つのバケットに入ります。ブロードキャストは、射影（Projection）、フィルタリング、 信頼できる統計によって、片側がすべてのExecutorに安全に収まることが証明されている場合にのみ適しています。180 GBのディメンションを強制的に ブロードキャストするのは危険です。AQEは低侵襲な第一の選択肢です。明示的なソルティングは、AQEがトリガーされない、またはSLAに 届かない場合の安定したホットキーに適しています。集約処理では、ソルト付きの部分集約に続いて2回目のマージを行う手法がよく使用されます。
完全な検証によって回答を締めくくります。パフォーマンスの修正では、行数、ビジネス指標の合計金額、 不明なキー、不一致率、重複率が変更されていないことも証明する必要があります。実行時間を96分から40分に短縮したとしても、 正当性は証明されず、翌日に異なるキー分布になった場合にソリューションが耐えうるかも示されません。
回答前に明確にすべき質問
- ボトルネックはスキャン、シャッフル書き込み、またはシャッフル読み取り後のいずれにあるか？ スキャンタスクの不均衡は、巨大なファイルや分割不可ファイルに起因する場合があります。本問ではJoinシャッフル後に異常値が現れるため、キースキューの経路に達します。
- 32%はレコード数、圧縮バイト数、または処理コストのどれを指しているか？ 行幅が広い場合、高負荷なUDF、出力のファンアウトがある場合、行数が控えめに見えてもコストのスキューが発生することがあります。レコード数、バイト数、時間、出力行数を比較してください。
- ビジネスにとって
UNKNOWNは何を意味するか？ 不明な製品にディメンション属性が不要な場合は、メインのJoinから分離し、元のコントラクトに基づいてNULL属性を補完します。番兵（Sentinel）ディメンション行と一致させる必要がある場合は、Joinを維持したままホットスポットを分割します。 - 製品ディメンションにおいて
product_idは一意か？UNKNOWNのディメンション行が複数存在すると、ホットなファクトキーが多対多の出力爆発を引き起こします。チューニングの前に、不正なカーディナリティとパーティションスキューを切り離します。 - 有効になっているSparkバージョンとAQE設定は何か？ Environmentページと最終的なアダプティブプランをチェックして、スイッチ、しきい値、Joinタイプ、およびスキュー分割が実際に実行されたエビデンスを確認します。
- 180 GBは元のディメンションか、それとも射影後のJoin入力か？ キーと2つの属性に絞り込むことで安全にブロードキャストできる場合、ブロードキャストが双方のシャッフルよりも優れている可能性があります。ランタイム統計とExecutorメモリ予算によってそれを証明する必要があります。
- どの不変条件（Invariant）が同等の結果を定義するか？ 最低限、総行数、ユニークイベント数、加法的なビジネス指標、不明キーの行数、不一致の行数、および許容される重複セマンティクスを指定します。
- ホットキーは安定的かつ列挙可能か？ 少数の安定したキーはターゲットソルティングに適しています。変化するロングテールは、AQE、動的なホットキー検出、または上流でのセマンティックな修正に適しています。
30秒の回答フレームワーク
「SQL UIでJoinタスクのシャッフル読み取り量、スピル、GC、およびリトライ先を比較します。中央値1.1 GiBに対して720 GiBのタスクが存在し、 32%がUNKNOWNであることはキースキューを裏付けます。出力爆発を排除するためにディメンションの一意性も検証します。 まずAQEのSkew Joinによって大きなパーティションが分割されることを確認します。実行時間が依然として45分を超える場合は、 安定したevent_idを用いてUNKNOWNを決定論的にソルト処理し、番兵行のみを複製します。 最後に、行数と金額をベースラインと突合し、タスク入力の最大/中央値比率、ステージ時間、コストを比較します。」
ステップごとの詳細な回答
ステップ1：96分の原因をステージとタスクに帰属させる
イベントログを保持し、Spark History Serverを使用して同一データ日付の正常な実行と性能劣化した実行を比較します。 SQLクエリをステージの詳細までドリルダウンします。パーセンタイル値と最大タスク所要時間を記録し、シャッフル読み取りのレコード数と バイト数、シャッフルスピル、ピーク実行メモリ、GC時間、失敗理由、Executorと照合します。SQLプランにおいて、Left Joinの前に Exchange hashpartitioning(product_id, 2000)があるか確認し、最終的な物理Joinを特定して、アダプティブプランが完了したかどうかを判断します。
ここでは、別のExecutorでリトライした後も同じタスクが約720 GiBを読み取り続けており、大半のタスクは約1.1 GiBを 読み取っています。これは入力パーティションそのものに原因があることを示しています。遅いタスクの入力が通常通りで、 特定のExecutorでGCやディスク待機が高い場合は、まずノードを調査します。すべてのタスクで一様にスピルが発生している場合は、 全体的なパーティションサイズとリソース予算に着目します。出力行数が突如急増した場合は、ディメンションの重複とJoin条件を検査します。
ステップ2：キー分布とJoinカーディナリティで原因を立証する
本番環境で使用されているものとまったく同じフィルターおよびキー正規化を適用した後に、ホットキーを測定します。 trim、大文字小文字の正規化、coalesce、またはUDFによって複数の値が1つのキーに統合される可能性があるため、 元のカラムをそのまま調べるだけでは不十分です。非常に大きなテーブルでは、既存の統計、制御されたサンプリング、または 制限付きの集約を使用して、診断自体が無制限の重いジョブにならないようにします。以下のクエリは、ファクトテーブルに すでにpayload_bytesが含まれているか事前計算されていることを前提としており、行数とバイト数の両方のスキューを測定できます。 そのカラムがない場合は、ストレージ統計または制御されたシリアライズ推定値を使用します。クエリは必要な集計を表しています：
SELECT
COALESCE(product_id, '<NULL>') AS join_key,
COUNT(*) AS row_count,
SUM(payload_bytes) AS payload_bytes
FROM fact_events
WHERE event_date = DATE '2026-07-17'
GROUP BY COALESCE(product_id, '<NULL>')
ORDER BY row_count DESC
LIMIT 20;また、各product_idがディメンション内で最大1回しか出現しないことを証明し、Join前後のファクトイベント数を比較します。 このLeft Joinでは、ディメンションキーが一意であれば、一致しないイベントも含め、各ファクトイベントは正確に1行を生成します。 UNKNOWNがファクト行の32%を占め、ディメンションに番兵行が1つあり、出力が増殖していない場合、 ハッシュパーティショニングが単一の720 GiBパーティションの発生理由となります。
ステップ3：実行技術を選択する前にセマンティックな原因に対処する
上流のリリースがなぜイベントの32%をUNKNOWNにマッピングしているのかを調査します。これがリグレッションである場合は、 マッピングをロールバックまたは修正し、影響を受けるパーティションを再実行します。これにより、データ品質とパフォーマンスの双方が回復します。 これが有効なビジネス値である場合、実行レイヤーはその分布に対応する必要があります。
製品属性を必要としない有効な不明イベントは、別経路をとることができます。コールドキーを通常通りJoinし、既存の契約に基づいて 不明経路のディメンション属性をNULLで埋め、unionByNameで結合します。これにより、情報損失なしにシャッフルを排除できます。 UNKNOWNが番兵属性と一致する必要がある場合は、Joinを維持し、AQEまたはターゲットソルト処理で作業を分割します。 結果のセマンティクスによって分岐を決定します。パーティションを均一にすることは、データを破棄してよい理由にはなりません。
ステップ4：効果的で最もコストの低い救済策を選択する
まずAQEを確認します。Spark 4.2.0では、spark.sql.adaptive.enabledおよび spark.sql.adaptive.skewJoin.enabledはデフォルトで有効になっていますが、クラスタ、ジョブ、またはマネージドプラットフォームによって上書きされる場合があります。 Skew Join係数としきい値バイト数の両方が一致する必要があります。最終的なアダプティブプランでスキュー処理を検査し、 ステージメトリクスで大きなパーティションが分割されていることを確認します。互換性のあるSort-Merge Joinの場合、AQEは大きい側を 分割し、小さい側を複製できます。これは日々の変動する分布に適応しますが、シャッフルと複製のコストが増加する可能性があり、 論理的に誤った多対多のJoinを修復することはできません。
射影されたディメンションに信頼性の高い統計があり、真に小さい場合は、ファクト側がJoinキーでシャッフルされないように Broadcast Hash Joinを検討します。元の180 GBは通常のブロードキャスト予算を大幅に超えています。強制的なヒントはすべての Executorを枯渇させる可能性があります。射影バイト数、同時実行タスク数、Executorヒープ、およびブロードキャストの タイムアウトを総合的に判断します。
AQEがトリガーされない、または依然としてSLAに届かない場合は、安定したホットキーを手動でソルト処理します。次のコードは、 event_idが安定的かつ一意であり、product_idがディメンション内で一意であり、UNKNOWNのみを分割する必要があると仮定しています。 ホットなファクト行は決定論的に32個のソルトにマッピングされます。一致する番兵ディメンション行のみが32回複製されます。 コールドキーはソルト0のまま維持されるため、ディメンション全体が増殖することはありません。
from pyspark.sql import functions as F
SALT_BUCKETS = 32
HOT_KEYS = ["UNKNOWN"]
events_salted = events.withColumn(
"salt",
F.when(
F.col("product_id").isin(*HOT_KEYS),
F.pmod(F.xxhash64("event_id"), F.lit(SALT_BUCKETS)).cast("int"),
).otherwise(F.lit(0)),
)
salt_values = spark.range(SALT_BUCKETS).select(
F.col("id").cast("int").alias("salt")
)
products_hot = (
products.filter(F.col("product_id").isin(*HOT_KEYS))
.crossJoin(salt_values)
)
products_cold = (
products.filter(~F.col("product_id").isin(*HOT_KEYS))
.withColumn("salt", F.lit(0))
)
products_salted = products_cold.unionByName(products_hot)
result = (
events_salted.join(products_salted, ["product_id", "salt"], "left")
.drop("salt")
)この面接シナリオにおいて32は初期候補です。ホットパーティションのバイト数、目標タスクサイズ、利用可能な並列度、 および小さい側の複製コストからバケット数を導き出し、代表的なデータでテストします。バケットが少なすぎるとロングテールが残り、 多すぎるとスケジューリング、ファイル、複製のオーバーヘッドが増加します。単にrepartition(4000, "product_id")を実行するだけでは、 依然としてすべてのUNKNOWN行が1つのパーティションに配置されます。
groupBy(product_id)の場合、通常は複製するディメンションがありません。まず(product_id, salt)で部分集約を行い、 次にその部分結果をproduct_idでマージします。sum、count、min、maxのように、結合法則および交換法則を満たす マージによって安全に分解できる操作はこの手法に適しています。厳密なmedian、順序に依存する集約、 マージ不可能なUDF状態には異なるアルゴリズムが必要です。
ステップ5：正当性、パフォーマンス、コストを1つの受け入れ基準に統合する
同一の不変な入力スナップショットに対してベースラインと候補を実行します。正当性が最優先です。総出力行数、 ユニークなevent_id、UNKNOWNの行数、不一致行数、および意味のあるディメンションごとのビジネス合計値とカウントを比較します。 ホットキー、コールドキー、NULL、および重複ディメンションキーについて行レベルの差分を抽出します。1つのファクトイベントが 1つのLeft Join行を生成するという保証はディメンションキーの一意性に依存するため、その制約は個別に監視します。
パフォーマンスについては、Joinステージにおけるp50、p95、最大タスク所要時間、シャッフル読み取りの最大/中央値比率、 スピル、GC、OOM、タスクリトライ、ステージ時間、総実行時間を比較します。コストについては、Executor時間（コア時間）、 シャッフルバイト数、出力ファイル数を記録します。45分以内に完了することは1つの基準にすぎません。シャッフルを倍増させたり、 結果を変更したり、翌日の新しいホットスポットで失敗したりしてSLAを達成しても、受け入れられません。
1つの過去日付でリプレイし、続いて新しいデータ日付でシャドウ実行して結果を比較することでリリースします。ホットキーセット、 タスク入力の最大/中央値比率、および不明キーの割合を監視します。上流のリリース後にUNKNOWNが32%に急増した事象は、 データ品質アラートをトリガーし、ジョブが遅延する前にセマンティックなリグレッションを露見させるべきです。
高品質な回答例
「まずSQL UIでリグレッションの原因を特定のJoinステージに帰属させます。現在のエビデンスはスキューを強く示唆しています。 2,000タスク全体でシャッフル読み取りの中央値が1.1 GiBであるのに対し、1つのタスクが720 GiBを読み取っており、そのタスクは 別のExecutorに移動しても遅いままです。最終的なアダプティブプラン、スピル、GCを調査し、本番とまったく同じ正規化を適用した後に キーをプロファイリングします。また、ディメンションキーの一意性も検証します。UNKNOWNのディメンション行が複数ある場合、 症状にJoin出力の爆発が含まれていることを意味します。
ディメンションが一意で、ファクト行の32%がUNKNOWNであると仮定すると、単一のハッシュシャッフルバケットが遅延タスクの 原因を説明します。パーティションを増やしてもバケットが増えるだけでそのキーは分割されず、Executorメモリを増やしてもOOMを先送りするだけです。 まず上流のマッピングがリグレッションかどうかを判断します。不明な行にディメンション属性が不要な場合は、それらをJoinから切り離し、 元のLeft Joinセマンティクスに従ってNULL属性を補完します。番兵行と一致させる必要がある場合は、ランタイム統計を使用して スキューしたSort-Merge-Joinパーティションを分割し、小さい側を複製できるAQE Skew Joinが最終プランに実際に現れているか確認します。
AQEを使用しても実行時間が45分を超える場合は、ターゲットソルティングを使用します。安定したevent_idによって、 各UNKNOWNファクト行を決定論的に例えば32個のソルトのいずれかに割り当てます。ディメンションのUNKNOWN行のみを それら32個のソルトに複製し、すべてのコールドキーはソルト0を使用します。各イベントは依然として1つのディメンション行に一致し、 複数のタスクがホットキーの作業を分担します。測定なしに32をハードコードするのではなく、ホットスポットのバイト数と目標タスクサイズから 最終的なバケット数を導き出します。
検証のために、同一の不変な入力をベースラインと比較します。総行数、ユニークイベント数、不明および不一致のレコード数、 ビジネスの合計値が一致する必要があります。その後、シャッフル読み取りの最大/中央値比率、タスク所要時間のテール、スピル、 OOM、ステージ時間、Executor時間、出力ファイルを比較します。最後に、過去の1日分をリプレイし、新しい1日分をシャドウ実行し、 不明キーの割合と新たなホットスポットを監視してアラートを設定します。これにより、ジョブが45分を満たし、結果を保持し、 上流の分布が再び変化した際にも可観測性を維持できることが証明されます。」
よくある間違い
- シャッフルパーティションを直ちに2,000から8,000に増やす → 他のタスクが小さくなるだけで、1つのホットキーは依然として1つのパーティションにハッシュされる → キー分布を測定し、AQE、セマンティックな分岐、またはターゲットソルティングでキーを分割する。
- Executorメモリのみを増やす → 1つのタスクの許容値は上がるが、720 GiBの作業量とロングテールはそのまま残る → まず最大パーティションのワークロードを削減し、測定されたタスクに基づいてリソースをサイジングする。
- 1つの遅いタスクを見ただけでスキューと断定する → 不良ノード、GC、リモートフェッチ、遅いUDFもStragglerの原因になり得る → タスク入力、リトライ先、スピル、GC、および実行計画を比較する。
- ファクトとディメンションを個別にランダムソルト処理する → ソルトが一致しなくなってJoin結果が失われ、リトライが非決定論的になる可能性がある → 安定した行IDからファクトのソルトを導き出し、ディメンション側で同じソルトを列挙する。
- ディメンション全体を32回複製する → 180 GBのディメンションでは膨大なネットワークおよびメモリコストが発生する → 確認されたホットキーの行のみを複製し、コールドキーはソルト0のままにする。
- 180 GBのディメンションを強制的にブロードキャストする → すべてのExecutorがブロードキャストデータを保持する必要があり、OOMで失敗する可能性がある → まず射影して測定する。メモリおよび同時実行の予算から安全であることが証明された後にのみブロードキャストする。
- ジョブを高速化するために
UNKNOWNをフィルタリングして除外する → 出力セマンティクスと下流のメトリクスが変化してしまう → 不明イベントの契約を確立し、分岐する場合でもLeft Joinの結果を保持する。 - 合計実行時間のみを比較する → 見かけ上の高速化は、データの欠落、重複、または誤計算によるものである可能性がある → タスク分散、コスト、SLAを比較する前に、行数およびビジネスの不変条件を証明する。
フォローアップの質問と回答方法
AQEが有効になっているにもかかわらず、なぜスキューしたJoinが分割されなかったのか？
最終的なアダプティブプランと、spark.sql.adaptive.enabled、Skew Joinスイッチ、 中央値係数のしきい値、および絶対バイト数のしきい値の有効な設定を検査します。両方のしきい値が一致する必要があります。 物理JoinがAQEでサポートされているパスに従っていること、ランタイム統計が利用可能であること、ヒントやプラットフォームによる上書きが 計画を制約していないことを確認します。追加のシャッフルを測定しながら、代表的なデータでしきい値の変更や強制的なスキュー最適化をテストします。 計画が恩恵を受けられない場合は、ターゲットソルティングを使用します。設定ファイルにtrue値があるからといって、 実行された計画がパーティションを分割したことの証明にはなりません。
射影によってディメンションが6 GiBに削減された場合、ブロードキャストできますか？
6 GiBであっても、Executorヒープ、同時実行タスク数、シリアライズサイズ、ブロードキャストタイムアウト、 クラスタの安定性の評価が必要です。ブロードキャストにより、大きい側のJoinキーのシャッフルを排除してホットスポットを回避できますが、 ディメンションが各Executorにコピーされます。統計情報を使用して実際のバイト数を証明し、ヒントを追加する前に本番相当の負荷テストで ピークメモリとGCを観察してください。「ファクトテーブルよりはるかに小さい」ことは、十分なブロードキャスト基準ではありません。
ホットキーが毎日変化し、HOT_KEYSを手動で維持できない場合はどうしますか？
AQEのランタイム応答を優先します。明示的なソルティングが依然として必要な場合は、メインジョブの前にレコード数、バイト数、または コストのしきい値を使用して制限付きのホットキーテーブルを作成し、その小さなテーブルをブロードキャストしてソルトパスを選択します。 データ日付ごとにリストをバージョン管理し、しきい値、件数制限、フォールバックを設定します。これには計画ステージと 運用状態が追加されるため、複雑さに見合うだけの安定したSLA向上が得られる必要があります。
遅い操作がgroupByである場合も、ディメンションを複製しますか？
いいえ。マージ可能な集約の場合、ホットな行にソルトを付与し、(key, salt)で部分集約を計算し、 それらの部分結果をkeyでマージします。これにより、1つのホットキーの入力が複数のタスクに分散され、 第2ステージは少数の部分結果を処理します。集約が安全にマージ可能であるかどうかを明記してください。全体で順序付けされた状態や マージ不可能な状態では、別のアルゴリズムを用いない限りこの手法は使用できません。
32個のソルトバケットをどのように選択しますか？
ホットパーティションのバイト数を目標タスク入力で割って下限を求め、利用可能なコア数、小さい側の複製コスト、 スケジューラオーバーヘッド、出力ファイルの制約を考慮します。720 GiBをタスクあたり約32 GiBまで削減する場合、 理論上の下限はおよそ23であるため、このシナリオでは32が妥当な実験値となります。最大タスク入力、ステージ時間、 Executor時間に関して複数の候補を比較し、ホットスポットの増加に備えて適度なヘッドルームを確保します。
スペキュレイティブ実行（推測実行）でこの遅延タスクを解決できますか？
決定論的なスキュータスクの複製は、依然として同じ720 GiBのパーティションを読み取るため、通常は2つのExecutorで高負荷な作業を 繰り返すだけになります。投機的実行は、断続的に遅いノードや一時的なジッターに対してより有用です。別の場所でリトライした後も 同じタスクが遅いかどうかを確認します。入力が外れ値のままである場合は、作業を複製するのではなく、データ処理を分割します。