デルタテーブルの圧縮

デルタ テーブル ファイルは、時間の経過と同時に断片化されます。 断片化により、ファイル操作のオーバーヘッドが増加し、圧縮効率が低下し、リーダーの並列処理が制限される可能性があります。 圧縮により、多数の小さなファイルが適切なサイズのファイルに書き換わるため、Spark はデータの読み取りと処理をより効率的に行うことができます。

OPTIMIZE コマンドは、主要な圧縮操作です。 理想的なファイル サイズをターゲットとするビンに小さなファイルをグループ化し、storageに書き換えます。

SQL Analytics エンドポイント、Power BI Direct Lake、Spark 全体の圧縮戦略に関するワークロード間のガイダンスについては、 ワークロード間のテーブルのメンテナンスと最適化に関するページを参照してください。

圧縮方法

Microsoft Fabricには、Delta テーブルで最適なファイル サイズを維持するためのいくつかの方法が用意されています。

OPTIMIZE コマンド

OPTIMIZE コマンドは、Delta テーブルを圧縮するための基本的な操作です。 デルタ テーブルのデータ レイアウトを向上させるために、小さなファイルを大きなファイルに書き換えます。

OPTIMIZE dbo.table_name

from delta.tables import *
deltaTable = DeltaTable.forName(spark, "dbo.table_name")
deltaTable.optimize().executeCompaction()

import io.delta.tables._
val deltaTable = DeltaTable.forName(spark, "dbo.table_name")
deltaTable.optimize().executeCompaction() 

OPTIMIZE はべき等ですが、特大の minFileSize では書き込み増幅が増加する可能性があります。 たとえば、 minFileSize 1 GB に設定すると、小さな余分な書き込みの後に 900 MB のファイルが書き換えられる場合があります。 ファイル サイズの自動管理ガイダンスについては、 アダプティブ ターゲット ファイル サイズに関するページを参照してください。

OPTIMIZE Z オーダー付き

ZORDER BY句を使用すると、OPTIMIZEはアクティブ なファイルを書き換えて、同じファイル内に類似した値を持つ行が併置されるようにします。 併置されたレイアウトにより、選択的フィルターのファイル スキップが改善されます。 Z オーダーは次の場合に使用します。

• クエリでは、多くの場合、2 つ以上の列 (日付 + customer_idなど) をフィルター処理します。
• これらの述語は十分に選択的であるため、ファイル レベルのスキップがスキャンされるファイルの数を減少させます。

OPTIMIZE dbo.table_name ZORDER BY (column1, column2)

OPTIMIZE V-オーダー付き

VORDER句を指定すると、圧縮対象のファイルに V オーダー最適化が適用されます。 V オーダーの詳細については、詳細な ドキュメントを参照してください。

OPTIMIZE dbo.table_name VORDER

1 つのコマンドで Z オーダーと V オーダーを組み合わせることができます。 Spark は、ビン圧縮→ Z オーダー → V オーダーの順序で操作を適用します。

OPTIMIZE dbo.table_name ZORDER BY (column1, column2) VORDER

OPTIMIZE中の V オーダー動作は、コマンドの呼び出し方法によって異なります。

OPTIMIZE 液体クラスタリングを搭載

液体クラスタリングはテーブルオプションとして指定されています。詳細については、 液体クラスタリングの有効化 を参照してください。 液体クラスタリングが有効になっている場合、 OPTIMIZE はクラスタリング ポリシーを適用する物理書き換えを実行します。

高速最適化

高速最適化では、Delta テーブル ファイルをインテリジェントに分析し、パフォーマンスを有意に向上させる可能性が高くない圧縮操作をスキップします。

小さなファイルが存在するたびにファイルを盲目的に圧縮する代わりに、高速最適化では、各候補ビン (小さなファイルのグループ) が構成可能なベスト プラクティスの圧縮目標を満たしているかどうかを評価します。 高速最適化では、マージがターゲットの最小サイズに達する可能性がある場合、または小さなファイルが多すぎる場合にのみ、ファイルのビンに圧縮が実行されます。 それ以外の場合は、そのグループをスキップするか、圧縮するファイルの数を減らします。

SET spark.microsoft.delta.optimize.fast.enabled = TRUE

spark.conf.set('spark.microsoft.delta.optimize.fast.enabled', True)

spark.conf.set("spark.microsoft.delta.optimize.fast.enabled", "true")

高速最適化は、圧縮の期待に基づいて微調整できます。

動作方法

高速最適化では、ビンを圧縮する前に追加のチェックが行われます。 候補のビンごとに、高速最適化によって次の結果が評価されます。

• ビン内の生データの推定量 (小さなファイル サイズの合計)
• 小さいファイルを組み合わせて、構成された最小ターゲット サイズを満たすファイルを生成するかどうか
• ビンに構成済みの最小数以上の小さいファイルが含まれているかどうか

高速最適化では、小さなファイルの各ビンが評価され、最小ターゲット サイズに達する可能性が高いビンまたは最小ファイル数を超える可能性のあるビンのみが圧縮されます。 これらのしきい値を満たしていないビンはスキップされるか、部分的に圧縮されます。 最適でないバケットをスキップすると、不要な書き換えが減り、書き込み負荷の増加を抑え、OPTIMIZE ジョブがより安定したものになります。

高速最適化を使用すると、Delta テーブルのライフサイクル全体で書き換えられたデータを削減できます。 次の図に示すように、高速最適化では最適でないビンがスキップされるため、書き込みの増幅が少なく、より高速で、よりべき等な OPTIMIZE ジョブになります。

制限事項

• 液体クラスタリングおよび Z オーダー操作には適用されません
• 高速最適化では、自動圧縮の動作は変更されません

ファイル レベルの圧縮ターゲット

圧縮最小サイズターゲットと最大ファイル サイズ ターゲットの変更に基づいて、以前に圧縮 (十分な大きさ) と見なされていたデータの書き換えを回避するために、 spark.microsoft.delta.optimize.fileLevelTarget.enabled を有効にして、既に圧縮されたファイルの再コンパイルを防ぐことができます。 有効にすると、ファイルは圧縮時にターゲット ファイル サイズの少なくとも半分に達していた場合、再圧縮されません。 ファイル レベルのターゲットを維持すると、圧縮ターゲット サイズが時間の経過と同時に変化する (たとえば、アダプティブ ターゲット ファイル サイズの評価とより大きなターゲットの設定から) ため、書き込み増幅が最小限に抑えられます。 有効にすると、OPTIMIZE の実行時に OPTIMIZE_TARGET_SIZE タグが新しいファイルに追加されます。または、 delta.targetFileSize または delta.targetFileSize.adaptive テーブル プロパティが設定されている場合は、書き込み操作に対して追加されます。

SET spark.microsoft.delta.optimize.fileLevelTarget.enabled = TRUE

spark.conf.set('spark.microsoft.delta.optimize.fileLevelTarget.enabled', True)

spark.conf.set("spark.microsoft.delta.optimize.fileLevelTarget.enabled", "true")

自動圧縮

自動圧縮では、書き込み操作のたびにパーティションの正常性が評価されます。 パーティション内の過剰なファイル断片化 (小さなファイルが多すぎる) を検出すると、書き込みがコミットされた直後に同期 OPTIMIZE 操作がトリガーされます。 圧縮はプログラムによって有益であると判断された場合にのみ実行されるため、ファイルのメンテナンスに対するこのライター駆動型のアプローチが最適です。

セッション レベルで有効にする

セッション レベルで spark.databricks.delta.autoCompact.enabled を設定して、その Spark セッションで作成された新しいテーブルの自動圧縮を有効にします。

SET spark.databricks.delta.autoCompact.enabled = TRUE

spark.conf.set('spark.databricks.delta.autoCompact.enabled', True)

spark.conf.set("spark.databricks.delta.autoCompact.enabled", "true")

テーブル レベルで有効にする

特定のテーブルの自動圧縮を有効にするには、table プロパティの delta.autoOptimize.autoCompact を設定します。

CREATE TABLE dbo.table_name
TBLPROPERTIES ('delta.autoOptimize.autoCompact' = 'true')

DataFrameWriter オプション delta.autoOptimize.autoCompact を使用して、テーブルの作成時に自動圧縮を有効にします。

df.write.option('delta.autoOptimize.autoCompact', 'true').saveAsTable('dbo.table_name')

既存のテーブルで同じテーブル プロパティを有効にします。

ALTER TABLE dbo.table_name
SET TBLPROPERTIES ('delta.autoOptimize.autoCompact' = 'true')

評価オーバーヘッドの削減

Fabric Spark ランタイム 2.0 (デルタ 4.1) 以降では、onCheckpointOnly 自動圧縮モードを有効にすることができます。 既定では、自動圧縮では、書き込み操作のたびにファイル メタデータが評価され、テーブルに小さなファイルが多すぎるかどうかを判断します。 onCheckpointOnly では、評価はログのチェックポイント処理まで延期されます（通常は 10 コミットごと）。 チェックポイント時には、テーブル スナップショットは既に完全に再構築されているため、評価では、余分なスキャンを必要とするのではなく、既にメモリ内にあるメタデータから読み取ります。 遅延評価により、コミットごとのオーバーヘッドを削減しつつ、テーブルが定期的にコンパクションされることが確実にされます。

SET spark.microsoft.delta.autoCompact.onCheckpointOnly.enabled = TRUE

spark.conf.set('spark.microsoft.delta.autoCompact.onCheckpointOnly.enabled', True)

spark.conf.set("spark.microsoft.delta.autoCompact.onCheckpointOnly.enabled", "true")