パフォーマンスの高いdatabricksSQL作成術：データアナリスト・サイエンティスト向けガイド

## はじめに

本記事はデータアナリストやデータサイエンティストを対象に、パフォーマンスの高いSQLの作成方法をまとめたものだ。Databricks SQLではオプティマイザがクエリを解析し最適化してくれるためパフォーマンスを考慮した書き方をしなくても良い一面もあるが、以下の点を意識することでさらなる効率化が可能となる。

## SQLの書き方

### 1. 必要な列の選択

- `SELECT *` は避け、必要な列のみを明示的に指定する。
これにより読み込むカラム数を減らし、I/O コストを抑制できる。
- 列数が多い場合は、ビューやサブクエリで中間的に絞り込んでからメインクエリを実行すると、処理効率が向上する。

### 2. パーティション化

どの列がパーティション化されているかは分析現場によって異なるので、データエンジニアに聞いてみるのが良い。

- 日付や地域などよく絞り込む列でテーブルをパーティション化し、クエリ実行時のデータスキャン量を大幅に削減する。
- パーティション数は多すぎても少なすぎても非効率なので、各パーティションあたり概ね 1 GB 以上を目安に設計する。

### 3. 結合の最適化（適切なJOIN戦略）

データ結合時は**小さいテーブルをブロードキャスト結合**するなど、ヒント句で最適戦略を指示できる（例：`/*+ BROADCAST(t) */`）。
Sparkの**AQE**を有効化すれば実行時に結合戦略を自動調整してくれる。
不要なシャッフルを避け、**結合処理を並列化**することで、特に小規模テーブルとの結合で大幅な速度向上が見込める。処理時間短縮により**クラスタ稼働時間を短く**できる。
AQEにより最適資源で実行し、無駄なコストを防げる。

```python
-- ヒント句を使用してsales_small_tableをブロードキャスト
SELECT /*+ BROADCAST(s) */ 
    c.customer_id,
    c.customer_name,
    s.product_id,
    s.sale_amount
FROM customer_large_table c
JOIN sales_small_table s ON c.customer_id = s.customer_id
WHERE c.region = 'APAC'
```

### 4. テーブル統計とCBOの活用

Deltaテーブルに対して**統計情報（ANALYZE）を更新**し、Sparkの**コストベースオプティマイザ(CBO)に最新情報を提供する。これにより結合順序や実行計画が自動的に改善される。
プランナーが最適な結合順序やインデックス利用**を選択しやすくなり、複雑なクエリでも高速化が期待できる。最適化されたプランで**無駄な計算を減らし**、結果的にクラスター稼働時間・リソース消費を抑制できる。

```python
-- 1. 最初に統計情報を更新
ANALYZE TABLE transactions COMPUTE STATISTICS FOR COLUMNS transaction_date, customer_id, product_id;
ANALYZE TABLE customers COMPUTE STATISTICS FOR COLUMNS customer_id, customer_segment;
ANALYZE TABLE products COMPUTE STATISTICS FOR COLUMNS product_id, category_id;

-- 2. CBOを有効化
SET spark.sql.cbo.enabled = true;
SET spark.sql.cbo.joinReorder.enabled = true;

-- 3. 最適化されたクエリ実行
SELECT 
    p.category_name,
    c.customer_segment,
    SUM(t.amount) as total_sales,
    COUNT(DISTINCT t.customer_id) as unique_customers
FROM 
    transactions t
JOIN 
    customers c ON t.customer_id = c.customer_id
JOIN 
    products p ON t.product_id = p.product_id
WHERE 
    t.transaction_date BETWEEN '2023-01-01' AND '2023-12-31'
GROUP BY 
    p.category_name,
    c.customer_segment
ORDER BY 
    total_sales DESC
```

### 5. 集約処理の工夫（必要に応じて概算）

巨大データに対する集計は**サンプリングや近似アルゴリズム**の利用も検討する。
Sparkは`APPROX_COUNT_DISTINCT`など**近似集計**をサポートし、精度を許容できれば計算コストを削減可能だ。粗い分析では**結果取得が速く**なる（正確な計算を避ける分、待ち時間を短縮）。
また、**CPU使用量やメモリ負荷を抑え**られ、大規模データ集計に伴うクラウドコストを軽減できる。

```python
-- 正確なDISTINCT COUNT（処理が重い）
SELECT 
    product_category,
    COUNT(DISTINCT customer_id) as exact_unique_customers
FROM sales
GROUP BY product_category;

-- 近似DISTINCT COUNT（処理が軽い）
SELECT 
    product_category,
    APPROX_COUNT_DISTINCT(customer_id) as approx_unique_customers
FROM sales
GROUP BY product_category;
```

## クエリ最適化テクニック一覧

| クエリ最適化テクニック | 実践的なヒント・説明 | パフォーマンス向上効果 | コスト削減効果 |
| --- | --- | --- | --- |
| **必要な列の選択** | クエリでは `SELECT *` を避け、**必要な列のみを選択**する。Delta Lake (Parquet) は**列指向**フォーマットのため、不要な列を読み飛ばせる。 | 読み込むデータ量を削減し、**I/Oやメモリ消費を減少**。結果としてクエリが高速化する。 | 読み込みスキャン量削減により、**計算リソース消費やDBU利用を削減**できる。 |
| **パーティションプルーニング** | テーブルがパーティション分割されている場合は、WHERE句に**パーティション列を含めて**データを絞る。Databricksはパーティション列だけでなく、関連するmin/max情報で**効率的に対象ファイルを絞り込み**する。 | 対象パーティション以外を読み込まなくなるため、**大幅なI/O削減**とレイテンシ短縮が得られる。 | スキャンするストレージ容量が減るため、その分**処理コスト(DBU含む)も削減**される。 |
| **結合の最適化** | データ結合時は**小さいテーブルをブロードキャスト結合**するなど、ヒント句で最適戦略を指示できる（例：`/*+ BROADCAST(t) */`）。Sparkの**AQE**を有効化すれば実行時に結合戦略を自動調整してくれる。 | 不要なシャッフルを避け、**結合処理を並列化**。特に小規模テーブルとの結合で大幅な速度向上が見込める。 | 処理時間短縮により**クラスタ稼働時間を短く**できる。AQEにより最適資源で実行し、無駄なコストを防ぐ。 |
| **テーブル統計とCBOの活用** | Deltaテーブルに対して**統計情報（ANALYZE）を更新**し、Sparkの**コストベースオプティマイザ(CBO)**に最新情報を提供する。これにより結合順序や実行計画が自動的に改善される。 | プランナーが**最適な結合順序やインデックス利用**を選択しやすくなり、複雑なクエリでも高速化が期待できる。 | 最適化されたプランで**無駄な計算を減らし**、結果的にクラスター稼働時間・リソース消費を抑制する。 |
| **集約処理の工夫** | 巨大データに対する集計は**サンプリングや近似アルゴリズム**の利用も検討する。Sparkは`APPROX_COUNT_DISTINCT`など**近似集計**をサポートし、精度を許容できれば計算コストを削減可能だ。 | 粗い分析では**結果取得が速く**なる（正確な計算を避ける分、待ち時間を短縮）。 | **CPU使用量やメモリ負荷を抑え**られ、大規模データ集計に伴うクラウドコストを軽減できる。 |

## おまけ：データ基盤を最適化する方法

SQLの書き方を工夫する以外にもデータ基盤を最適化することによってパフォーマンスを向上させることも可能だ。データエンジニアの人は詳しく調べて見るのが良い。

### デルタキャッシュ

Databricksクラスタ起動時に以下の設定を追加すると自動的にディスクキャッシュが動作する：

```python
spark.conf.set("spark.databricks.io.cache.enabled", "true")
```

### 概要

- ディスクキャッシュはリモートストレージ（S3/ABFSなど）のParquetファイル（Delta Lake含む）をワーカーノードのローカルSSDにコピーし、高速な中間フォーマットでキャッシュする。
- キャッシュはファイル読み込み時に自動的に適用され、2回目以降の読み込みはローカルストレージから実行されるためI/Oが大幅に削減される。

## まとめ

以上、Databricks SQL 上でDelta Lakeの大規模データを扱う際のポイントを整理した。
**クエリの工夫＋データレイク最適化＋リソース管理**を組み合わせることで、応答時間短縮と費用対効果の高い分析基盤を実現できる。