Databricks Query Optimization Patterns: ファイルサイズ・AQE・Join最適化・Spill対策

Databricksでのクエリパフォーマンスは「ファイルサイズの適正化」「AQE（Adaptive Query Execution）の活用」「Join戦略の選択」「Spill（ディスク溢出）の回避」の4軸で大きく改善します。 本稿では各軸の設計パターンとパラメータ設定を、Data Engineer Professional試験と実務の両面から整理します。

ファイルサイズ設計

Delta LakeやParquetのファイルサイズは、読み取りパフォーマンスに直結します。 ファイルが小さすぎるとメタデータオーバーヘッドとタスク起動コストが増大し（Small File Problem）、 大きすぎるとパーティション内の並列度が下がります。

-- OPTIMIZEの実行（ZORDERと組み合わせ可能）
OPTIMIZE gold.orders ZORDER BY (customer_id, order_date);

-- テーブルプロパティでターゲットサイズを設定
ALTER TABLE gold.orders
SET TBLPROPERTIES (
  'delta.targetFileSize' = '134217728'  -- 128MB
);

-- autoOptimizeの有効化
ALTER TABLE silver.events
SET TBLPROPERTIES (
  'delta.autoOptimize.optimizeWrite' = 'true',
  'delta.autoOptimize.autoCompact' = 'true'
);

ZORDER BYはフィルタ条件に頻出するカラムを指定します。 ZORDERはカラム値でデータを空間的にクラスタリングし、Data Skippingの効果を最大化します。 ただし、ZORDERは高カーディナリティのカラム（UUIDなど）には効果が薄いため、3〜4カラム以下に絞ります。

AQE（Adaptive Query Execution）設定パラメータ

AQEは実行時の統計情報に基づいてクエリプランを動的に最適化する機能です。 Databricksではデフォルトで有効であり、以下の3つの最適化を自動的に行います。

• Coalescing Post-Shuffle Partitions: Shuffleの結果パーティションが小さすぎる場合に結合
• Converting Sort-Merge Join to Broadcast Hash Join: 実行時にテーブルサイズが十分小さいと判明した場合にBroadcast Joinに切り替え
• Optimizing Skew Joins: 特定のキーに偏ったパーティション（データスキュー）を分割して並列化

Join最適化: 戦略の選択基準

SparkのJoin戦略は主に3種類あり、テーブルサイズとデータ特性に応じて選択されます。 AQEが有効な場合は実行時に動的に切り替わりますが、設計段階での理解が重要です。

-- 小テーブルを明示的にBroadcastする（ヒント指定）
SELECT /*+ BROADCAST(dim_products) */
  f.order_id,
  f.amount,
  d.product_name
FROM fact_orders f
JOIN dim_products d ON f.product_id = d.product_id;

-- autoBroadcastJoinThresholdの調整
SET spark.sql.autoBroadcastJoinThreshold = 50m;  -- 50MBまでBroadcast

BROADCASTヒントはオプティマイザの判断を上書きします。 ディメンションテーブル（数万〜数十万行）とファクトテーブル（数億行）のJoinでは、ディメンション側にBROADCASTヒントを指定するのが定石です。

Shuffleパーティション数の設計

spark.sql.shuffle.partitionsはShuffle時のパーティション数を制御します。デフォルトは200ですが、データ量に応じて調整が必要です。

• データ量が小さい場合（数GB以下）: 200は過大。パーティション数を減らすことでタスクオーバーヘッドを削減
• データ量が大きい場合（数百GB以上）: 200では各パーティションが大きすぎてSpillが発生。パーティション数を増やす
• AQEが有効なら: coalescePartitionsが自動で小パーティションを結合するため、大きめに設定して問題ない

-- Shuffleパーティション数の調整
SET spark.sql.shuffle.partitions = 400;

-- AQE有効時の推奨パーティションサイズ
SET spark.sql.adaptive.advisoryPartitionSizeInBytes = 128m;

Spill（ディスク溢出）対策

SpillはShuffle、Sort、Aggregation時にメモリが不足した場合にデータをディスクに退避する動作です。 パフォーマンスが大幅に劣化するため、Spillの発生を最小化する設計が重要です。

-- Spark UIで確認すべきメトリクス
-- Stages > Task Details:
--   Shuffle Spill (Memory): メモリからディスクに溢れたデータ量
--   Shuffle Spill (Disk):   ディスクに書き込まれたデータ量

-- Spillが発生している場合の対策例
SET spark.sql.shuffle.partitions = 800;    -- パーティション分割を細かく
SET spark.sql.adaptive.advisoryPartitionSizeInBytes = 64m; -- AQE推奨サイズを小さく

Photonエンジンの活用

PhotonはDatabricksのネイティブベクトル化実行エンジンで、Scan、Filter、Aggregation、Joinの各ステージを高速化します。 Photon対応のクラスターを使用するだけで有効になり、特別なSQLの書き換えは不要です。

• Scan: Parquet/Deltaの列指向読み取りをSIMD命令で高速化
• Filter/Aggregation: C++ネイティブコードで処理し、JVMのGCオーバーヘッドを排除
• Join: ハッシュテーブル構築がネイティブコードで高速化
• 制約: UDFやPython UDFはPhoton対象外。SQL/DataFrameの組み込み関数を使うことでPhotonの恩恵を最大化

最適化チェックリスト

• テーブルに定期的にOPTIMIZE + ZORDERを実行し、ファイルサイズを128MB前後に維持する
• ストリーミングテーブルにはautoOptimizeを有効化する
• AQEはデフォルト有効のまま運用し、advisoryPartitionSizeInBytesをデータ量に応じて調整する
• ディメンションテーブルとのJoinではBROADCASTヒントを検討する
• Spark UIのStagesタブでShuffle Spill (Disk)を監視し、Spillが恒常的に発生するクエリは設計を見直す
• Photon対応クラスターを使用し、可能な限りSQL組み込み関数でロジックを書く

問題で確認

よくある質問