Delta テーブルに Liquid Clustering クラスタリングを使用する

[アーティクル]
04/23/2024

重要

Delta Lake リキッドクラスタリングは、Databricks Runtime 13.3 以降のパブリックプレビューで使用できます。リキッドクラスタリングは Databricks Runtime 12.2 LTS 以降で一部サポートされています。「リキッドクラスタリングありのテーブルの互換性」を参照してください。

Delta Lake リキッドクラスタリングでは、テーブルのパーティション分割と ZORDER が置き換えられ、データレイアウトの決定が簡略化され、クエリのパフォーマンスが最適化されます。リキッドクラスタリングを使用すると、既存のデータを書き換えることなくクラスタリングキーを柔軟に再定義できるため、時間の経過と共に分析ニーズに沿ってデータレイアウトを発展させることができます。

警告

リキッドクラスタリングを有効にしたDelta テーブルを作成、記述、OPTIMIZEするには、Databricks Runtime 13.3 LTS 以上が必要です。

Note

Databricks Runtime 13.3 LTS 以上では、リキッドクラスタリングが有効になっているテーブルでは、行レベルのコンカレンシーがサポートされます。 Databricks Runtime 14.2 以降では、削除ベクトルが有効になっているすべてのテーブルに対して、行レベルのコンカレンシーが一般提供されています。 Azure Databricks での分離レベルと書き込みの競合に関するページを参照してください。

リキッドクラスタリングの用途

Databricks では、すべての新しい Delta テーブルに対してリキッドクラスタリングをお勧めしています。クラスタリングのメリットが得られるシナリオの例を次に示します。

カーディナリティの高い列でフィルター処理されることが多いテーブル。
データ分散に大きな偏りがあるテーブル。
急速に大きくなり、メンテナンスとチューニング作業が必要なテーブル。
同時書き込みの要件を持つテーブル。
時間の経過と共に変化するアクセスパターンを持つテーブル。
一般的なパーティションキーでは、パーティションが多すぎるか、少なすぎるテーブルが残ってしまう可能性があるテーブル。

リキッドクラスタリングを有効にする

既存のテーブルで、またはテーブルの作成時に、リキッドクラスターを有効にすることができます。クラスタリングはパーティション分割または ZORDER と互換性がありません。また、テーブル内のデータに対するすべてのレイアウト操作と最適化操作を Azure Databricks クライアントで管理する必要があります。有効にしたら、OPTIMIZE ジョブを通常どおりに実行して、データを増分でクラスター化します。「クラスタリングをトリガーする方法」を参照してください。

リキッドクラスタリングを有効にするには、次の例のように、テーブル作成ステートメントに CLUSTER BY フレーズを追加します。

Note

Databricks Runtime 14.2 以上では、Python または Scala で DataFrame API と DeltaTable API を使用して、リキッドクラスタリングを有効にすることができます。

SQL

-- Create an empty table
CREATE TABLE table1(col0 int, col1 string) USING DELTA CLUSTER BY (col0);

-- Using a CTAS statement
CREATE EXTERNAL TABLE table2 CLUSTER BY (col0)  -- specify clustering after table name, not in subquery
LOCATION 'table_location'
AS SELECT * FROM table1;

-- Using a LIKE statement to copy configurations
CREATE TABLE table3 LIKE table1;

Python

# Create an empty table
(DeltaTable.create()
  .tableName("table1")
  .addColumn("col0", dataType = "INT")
  .addColumn("col1", dataType = "STRING")
  .clusterBy("col0")
  .execute())

# Using a CTAS statement
df = spark.read.table("table1")
df.write.format("delta").clusterBy("col0").saveAsTable("table2")

# CTAS using DataFrameWriterV2
df = spark.read.table("table1")
df.writeTo("table1").using("delta").clusterBy("col0").create()

Scala

// Create an empty table
DeltaTable.create()
  .tableName("table1")
  .addColumn("col0", dataType = "INT")
  .addColumn("col1", dataType = "STRING")
  .clusterBy("col0")
  .execute()

// Using a CTAS statement
val df = spark.read.table("table1")
df.write.format("delta").clusterBy("col0").saveAsTable("table2")

// CTAS using DataFrameWriterV2
val df = spark.read.table("table1")
df.writeTo("table1").using("delta").clusterBy("col0").create()

警告

リキッドクラスタリングを有効にして作成されたテーブルでは、作成時に多数の Delta テーブル機能が有効になり、Delta ライター (バージョン 7) とリーダー (バージョン 3) が使用されます。これらの機能の一部の有効化をオーバーライドできます。「既定の機能の有効化をオーバーライドする (省略可能)」を参照してください。

テーブルプロトコルのバージョンをダウングレードすることはできません。また、有効になっている Delta リーダープロトコルテーブルの機能がすべてサポートされていない Delta Lake クライアントでは、クラスタリングが有効になっているテーブルを読み取ることはできません。「Azure Databricks で Delta Lake 機能の互換性を管理する方法は?」を参照してください。

Databricks Runtime 13.3 LTS 以上においては、次の構文を使用して、既存のパーティション分割されていない Delta テーブル上でリキッドクラスタリングを有効にすることができます。

ALTER TABLE <table_name>
CLUSTER BY (<clustering_columns>)

既定の機能の有効化をオーバーライドする (省略可能)

リキッドクラスタリングの有効化中に Delta テーブル機能を有効にする既定の動作をオーバーライドできます。これにより、これらのテーブル機能に関連付けられているリーダープロトコルとライタープロトコルがアップグレードされなくなります。次の手順を実行するには、既存のテーブルが必要です。

ALTER TABLE を使用して、1 つ以上の機能を無効にするテーブルプロパティを設定します。たとえば、削除ベクトルを無効にするには、次を実行します。
```
ALTER TABLE table_name SET TBLPROPERTIES ('delta.enableDeletionVectors' = false);
```
次を実行して、テーブルでリキッドクラスタリングを有効にします。
```
ALTER TABLE <table_name>
CLUSTER BY (<clustering_columns>)
```

次の表に、オーバーライドできる Delta 機能と、有効化が Databricks Runtime バージョンとの互換性に与える影響に関する情報を示します。

Delta 機能	ランタイムの互換性	有効化をオーバーライドするプロパティ	リキッドクラスタリングに対する無効化の影響
削除ベクトル	読み取りと書き込みには、Databricks Runtime 12.2 LTS 以降が必要です。	`'delta.enableDeletionVectors' = false`	行レベルのコンカレンシーが無効になっているため、トランザクションとクラスタリング操作が競合する可能性が高くなります。「行レベルのコンカレンシーでの書き込みの競合」を参照してください。 `DELETE`、`MERGE`、および `UPDATE` コマンドの実行速度が低下する場合があります。
行追跡	書き込みには、Databricks Runtime 13.3 LTS 以降が必要です。任意の Databricks Runtime バージョンから読み取ることができます。	`'delta.enableRowTracking' = false`	行レベルのコンカレンシーが無効になっているため、トランザクションとクラスタリング操作が競合する可能性が高くなります。「行レベルのコンカレンシーでの書き込みの競合」を参照してください。
チェックポイント V2	読み取りと書き込みには、Databricks Runtime 13.3 LTS 以降が必要です。	`'delta.checkpointPolicy' = 'classic'`	リキッドクラスタリングの動作には影響ありません。

クラスタリングキーを選択する

Databricks では、一般的に使用されるクエリフィルターに基づいてクラスタリングキーを選択することをお勧めしています。クラスタリングキーは任意の順序で定義できます。 2 つの列が関連付けられた場合、必要なのは、いずれかの列をクラスタリングキーとして追加することだけです。

クラスタリングでは、クラスタリングキーについて次のデータ型がサポートされています。

日
Timestamp
TimestampNTZ (Databricks Runtime 14.3 LTS 以降が必要です)
String
Integer
Long
Short
Float
Double
10 進法
Byte
Boolean

既存のテーブルを変換する場合は、次の推奨事項を考慮してください。

現在のデータ最適化手法	クラスタリングキーの推奨事項
Hive スタイルのパーティション分割	パーティション列をクラスタリングキーとして使用します。
Z オーダーのインデックス作成	`ZORDER BY` 列をクラスタリングキーとして使用します。
Hive スタイルのパーティション分割と Z オーダー	パーティション列と `ZORDER BY` 列の両方をクラスタリングキーとして使用します。
カーディナリティを減らすために生成された列 (タイムスタンプの日付など)	元の列をクラスタリングキーとして使用し、生成された列を作成しないでください。

クラスター化されたテーブルにデータを書き込む

リキッドクラスタリングで使用されるすべての Delta 書き込みプロトコルテーブル機能がサポートされている Delta ライタークライアントを使用する必要があります。 Azure Databricks では、Databricks Runtime 13.3 LTS 以降を使用する必要があります。

ほとんどの操作では、書き込み時にデータは自動的にクラスター化されません。書き込み時にクラスター化される操作は次のとおりです。

INSERT INTO 操作
CTAS ステートメント
Parquet 形式からの COPY INTO
spark.write.format("delta").mode("append")

Note

書き込み時のクラスタリングはベストエフォートアプリケーションであり、次の状況では適用されません。

書き込み操作が 512 GB を超えるデータの場合。
SELECT サブクエリに変換、フィルター、または結合が含まれている場合。
投影列がソーステーブルと同じでない場合。

すべての操作にリキッドクラスタリングが適用されるわけではないため、Databricks では、すべてのデータが効率的にクラスター化されるように、OPTIMIZE を頻繁に実行することをお勧めします。

クラスタリングをトリガーする方法

クラスタリングをトリガーするには、Databricks Runtime 13.3 LTS 以降を使用する必要があります。次の例のように、テーブルで OPTIMIZE コマンドを使用します。

OPTIMIZE table_name;

リキッドクラスタリングは増分です。つまり、クラスター化する必要があるデータに対応するために、必要に応じてデータのみが書き換えられます。クラスター化するデータと一致しないクラスタリングキーを持つデータファイルは書き換えられません。

最適なパフォーマンスを得るために、Databricks では、通常の OPTIMIZE ジョブをクラスターデータにスケジュールすることをお勧めしています。多数の更新または挿入が発生しているテーブルの場合、Databricks では、1 時間または 2 時間ごとに OPTIMIZE ジョブをスケジュールすることをお勧めしています。リキッドクラスタリングは増分であるため、クラスター化されたテーブルのほとんどの OPTIMIZE ジョブは迅速に実行されます。

クラスター化されたテーブルからデータを読み取る

削除ベクトルの読み取りがサポートされている Delta Lake クライアントを使用して、クラスター化されたテーブル内のデータを読み取ることができます。最適なクエリ結果を得るには、次の例のように、クエリフィルターにクラスタリングキーを含めます。

SELECT * FROM table_name WHERE cluster_key_column_name = "some_value";

クラスタリングキーを変更する

テーブルのクラスタリングキーは、次の例のように、ALTER TABLE コマンドを実行することでいつでも変更できます。

ALTER TABLE table_name CLUSTER BY (new_column1, new_column2);

クラスタリングキーを変更すると、後続の OPTIMIZE 操作と書き込み操作では新しいクラスタリングアプローチが使用されますが、既存のデータは書き換えられません。

次の例のように、キーを NONE に設定してクラスタリングを無効にすることもできます。

ALTER TABLE table_name CLUSTER BY NONE;

クラスターキーを NONE に設定しても、既にクラスター化されているデータは書き換えられませんが、今後の OPTIMIZE 操作ではクラスタリングキーを使用できなくなります。

テーブルがクラスター化される方法を確認する

次の例のように、DESCRIBE コマンドを使用して、テーブルのクラスタリングキーを確認できます。

DESCRIBE TABLE table_name;

DESCRIBE DETAIL table_name;

リキッドクラスタリングありのテーブルの互換性

Databricks では、リキッドクラスタリングを有効にしているテーブルとの間で読み書きするあらゆるワークロードについて、Databricks Runtime 13.3 LTS 以降を推奨しています。

Databricks Runtime 14.1 以降でリキッドクラスタリングありで作成されたテーブルでは、既定で v2 チェックポイントが使用されます。 Databricks Runtime 13.3 LTS 以降では、v2 チェックポイントを使用してテーブルを読み書きできます。

Databricks Runtime 12.2 LTS 以降では、v2 チェックポイントを無効にし、テーブルプロトコルをダウングレードすることでリキッドクラスタリングありのテーブルを読み取ることができます。「Delta テーブル機能の削除」を参照してください。

制限事項

次の制限があります。

クラスタリングキーには、収集された統計を含む列のみを指定できます。既定では、Delta テーブルの最初の 32 列で統計が収集されます。
最大で 4 列をクラスタリングキーとして指定できます。
構造化ストリーミングワークロードでは、書き込み時のクラスタリングはサポートされていません。

Delta テーブルに Liquid Clustering クラスタリングを使用する

リキッド クラスタリングの用途

リキッド クラスタリングを有効にする

SQL

Python

Scala

既定の機能の有効化をオーバーライドする (省略可能)

クラスタリング キーを選択する

クラスター化されたテーブルにデータを書き込む

クラスタリングをトリガーする方法

クラスター化されたテーブルからデータを読み取る

クラスタリング キーを変更する

テーブルがクラスター化される方法を確認する

リキッド クラスタリングありのテーブルの互換性

制限事項

その他のリソース

リキッドクラスタリングの用途

リキッドクラスタリングを有効にする

クラスタリングキーを選択する

クラスタリングキーを変更する

リキッドクラスタリングありのテーブルの互換性