Databricksのクエリーヒント

[Hints \| Databricks on AWS](https://docs.databricks.com/spark/latest/spark-sql/language-manual/sql-ref-syntax-qry-select-hints.html) [2022/1/21時点]の翻訳です。

:::note warn
本書は抄訳であり内容の正確性を保証するものではありません。正確な内容に関しては原文を参照ください。
:::

実行計画を生成するために特定のアプローチを提案します。

# 構文

```
/*+ hint [, ...] */
```

# パーティショニングのヒント

パーティショニングのヒントを用いることで、Databricksランタイムが従うべきパーティショニング戦略を提案することができます。`COALESCE`、`REPARTITION`、`REPARTITION_BY_RANGE`ヒントがサポートされており、Dataset APIの`coalesce`、`repartition`、`repartitionByRange`と同じものとなります。これらのヒントを用いることで、パフォーマンスチューニングや出力ファイルのコントロールが可能となります。複数のパーティショニングヒントが指定された場合、論理的なプランに複数のノードが挿入されますが、オプティマイザは一番左のヒントを使用します。

## パーティショニングヒントのタイプ

- **COALESCE**
  指定されたパーティション数にパーティションの数を減らします。パラメーターとしてパーティション数を受け取ります。
- **REPARTITION**
  指定されたパーティション表現を用いて、指定されたパーティション数に再パーティショニングを行います。パーティション数やカラム名、あるいは両方をパラメーターとして受け取ります。
- **REPARTITION_BY_RANGE**
  指定されたパーティション表現を用いて、指定されたパーティション数に再パーティショニングを行います。パラメーターとしてカラム名、オプションとしてパーティション数を受け取ります。
- **REBALANCE**
  `REBALANCE`ヒントは、クエリー結果の出力パーティションのリバランスのために使用され、それぞれのパーティションが合理的なサイズ(小さすぎもなく大きすぎもなく)になります。パラメーターとしてカラム名を受け取り、これらのカラムを用いてクエリー結果をベストなパーティションになるようにしようとします。これはベストエフォートです。偏りがある場合、Sparkは偏っているパーティションを分割し、パーティションが大きくなりすぎないようにします。このヒントは、クエリーの結果をテール部に書き込む際に、大きすぎる・小さすぎるファイルを書き込むことを回避する役に立ちます。このヒントはAQE(Adaptive Query Execution)が無効化されている場合には無視されます。

## 例

```sql:SQL
> SELECT /*+ COALESCE(3) */ * FROM t;

> SELECT /*+ REPARTITION(3) */ * FROM t;

> SELECT /*+ REPARTITION(c) */ * FROM t;

> SELECT /*+ REPARTITION(3, c) */ * FROM t;

> SELECT /*+ REPARTITION_BY_RANGE(c) */ * FROM t;

> SELECT /*+ REPARTITION_BY_RANGE(3, c) */ * FROM t;

> SELECT /*+ REBALANCE */ * FROM t;

> SELECT /*+ REBALANCE(c) */ * FROM t;

-- multiple partitioning hints
> EXPLAIN EXTENDED SELECT /*+ REPARTITION(100), COALESCE(500), REPARTITION_BY_RANGE(3, c) */ * FROM t;
== Parsed Logical Plan ==
'UnresolvedHint REPARTITION, [100]
+- 'UnresolvedHint COALESCE, [500]
   +- 'UnresolvedHint REPARTITION_BY_RANGE, [3, 'c]
      +- 'Project [*]
         +- 'UnresolvedRelation [t]

== Analyzed Logical Plan ==
name: string, c: int
Repartition 100, true
+- Repartition 500, false
   +- RepartitionByExpression [c#30 ASC NULLS FIRST], 3
      +- Project [name#29, c#30]
         +- SubqueryAlias spark_catalog.default.t
            +- Relation[name#29,c#30] parquet

== Optimized Logical Plan ==
Repartition 100, true
+- Relation[name#29,c#30] parquet

== Physical Plan ==
Exchange RoundRobinPartitioning(100), false, [id=#121]
+- *(1) ColumnarToRow
   +- FileScan parquet default.t[name#29,c#30] Batched: true, DataFilters: [], Format: Parquet,
      Location: CatalogFileIndex[file:/spark/spark-warehouse/t], PartitionFilters: [],
      PushedFilters: [], ReadSchema: struct<name:string>
```

# Joinのヒント

Joinのヒントを用いることで、Databricksランタイムが使用すべきjoin戦略を提案することができます。joinの両側に異なるjoin戦略ヒントが指定された場合、Databricksは以下の順序でヒントの優先づけを行います。左の方が優先度が高くなります：`BROADCAST`、`MERGE`、`SHUFFLE_HASH`、`SHUFFLE_REPLICATE_NL`。joinの両側に`BROADCAST`あるいは`SHUFFLE_HASH`がある場合には、Databricksランタイムはjoinのタイプとリレーションのサイズに基づいて構築する側を選択します。指定される戦略では全てのjoinのタイプをサポートしていない場合があるので、Databricksランタイムはヒントによって提案されたjoin戦略を使用することを保証しません。

## Joinヒントのタイプ

- **BROADCAST**
  broadcast joinを使用します。このヒントが指定されたjoinの側は`autoBroadcastJoinThreshold`に関係なくブロードキャストされます。joinの両側にbroadcastヒントがある場合、サイズの小さい(統計情報を用います)側がブロードキャストされます。`BROADCAST`のエイリアスには、`BROADCASTJOIN`と`MAPJOIN`があります。
- **MERGE**
  shuffle sort merge joinを使用します。`MERGE`のエイリアスには`SHUFFLE_MERGE`と`MERGEJOIN`があります。
- **SHUFFLE_HASH**
  shuffle hash joinを使用します。両側にshuffle hashヒントがある場合、サイズの小さい(統計情報を用います)側を構築する側として選択します。
- **SHUFFLE_REPLICATE_NL**
  shuffle-and-replicate nested loop joinを使用します。

## 例

```sql:SQL
-- Join Hints for broadcast join
> SELECT /*+ BROADCAST(t1) */ * FROM t1 INNER JOIN t2 ON t1.key = t2.key;
> SELECT /*+ BROADCASTJOIN (t1) */ * FROM t1 left JOIN t2 ON t1.key = t2.key;
> SELECT /*+ MAPJOIN(t2) */ * FROM t1 right JOIN t2 ON t1.key = t2.key;

-- 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;

-- When different join strategy hints are specified on both sides of a join, Databricks Runtime
-- prioritizes the BROADCAST hint over the MERGE hint over the SHUFFLE_HASH hint
-- over the SHUFFLE_REPLICATE_NL hint.
-- Databricks Runtime will issue Warning in the following example
-- org.apache.spark.sql.catalyst.analysis.HintErrorLogger: Hint (strategy=merge)
-- is overridden by another hint and will not take effect.
SELECT /*+ BROADCAST(t1), MERGE(t1, t2) */ * FROM t1 INNER JOIN t2 ON t1.key = t2.key;
```

# Skewのヒント

(Delta Lake on Databricks) `SKEW`ヒントに関しては、[Skew join optimization](https://docs.databricks.com/delta/join-performance/skew-join.html)を参照ください。

# 関連コマンド

- [SELECT](https://docs.databricks.com/spark/latest/spark-sql/language-manual/sql-ref-syntax-qry-select.html)

### Databricks 無料トライアル

[Databricks 無料トライアル](https://databricks.com/jp/try-databricks)