テーブル履歴の操作

Apache Iceberg テーブルと Delta Lake テーブルの場合、テーブルを変更する各操作で新しいテーブル バージョンが作成されます。 履歴情報を使用して、操作の監査、テーブルのロールバック、タイム トラベルを使用した特定の時点でのテーブルのクエリを実行します。

テーブル履歴の取得

DESCRIBE HISTORY コマンドを実行して、テーブルへの書き込みごとに操作、ユーザー、タイムスタンプなどの情報を取得します。 操作は、逆の時系列順で返されます。

DESCRIBE HISTORY返す列、operationParameters列の値、operationMetrics列の操作ごとのメトリクスについては、「テーブル履歴スキーマおよびオペレーションメトリクス」を参照してください。

テーブル履歴の保持期間は、テーブル設定 logRetentionDuration によって決まり、既定では 30 日間です。

DESCRIBE HISTORY table_name       -- get the full history of the table

DESCRIBE HISTORY table_name LIMIT 1  -- get the last operation only

Spark SQL 構文の詳細については、DESCRIBE HISTORYを参照してください。

Scala、Java、およびPython構文の詳細については、Delta Lake API のドキュメントを参照してください。

カタログ エクスプローラーの[履歴 ] タブにテーブル履歴が視覚的に表示されます。

OPTIMIZE操作の種類を識別する

自動圧縮、液体クラスタリング、Z オーダーはすべて、テーブル履歴に OPTIMIZE 操作として表示されます。 どれが実行されたかを確認するには、operationParameters 列を調べます。

テーブルの履歴内のすべての OPTIMIZE 操作を分類するには、次のコマンドを実行します。

SELECT
  version,
  timestamp,
  CASE
    WHEN operationParameters.clusterBy IS NOT NULL AND operationParameters.clusterBy <> '[]' THEN 'Liquid clustering'
    WHEN operationParameters.zOrderBy IS NOT NULL AND operationParameters.zOrderBy <> '[]' THEN 'Z-ordering'
    WHEN operationParameters.auto = 'true' THEN 'Auto compaction'
    ELSE 'Manual OPTIMIZE'
  END AS optimize_type,
  operationParameters.auto AS is_auto_compaction,
  operationParameters.clusterBy AS cluster_by,
  operationParameters.zOrderBy AS z_order_by,
  operationMetrics.numRemovedFiles AS files_compacted,
  operationMetrics.numAddedFiles AS files_added,
  operationMetrics.numRemovedBytes AS bytes_removed,
  operationMetrics.numAddedBytes AS bytes_added
FROM (DESCRIBE HISTORY table_name)
WHERE operation = 'OPTIMIZE'
ORDER BY version DESC;

次のセクションでは、各 operationParameters 値について詳しく説明します。 前のクエリが選択する operationMetrics キーの定義については、 Operation metricsを参照してください。

自動圧縮

自動圧縮では、 auto パラメーターが trueに設定されます。 Azure Databricksは、書き込み後に自動圧縮を自動的にトリガーします。 autoがfalseされると、ユーザーまたはスケジュールされたジョブが OPTIMIZE コマンドを実行しました。

たとえば、自動圧縮操作では次の内容が示されます。

operationParameters: {
  "auto": "true"
}

自動圧縮の詳細については、「 自動圧縮」を参照してください。

リキッド クラスタリング

液体クラスタリングでは、 clusterBy パラメーターにクラスタリング列名が設定されます。 空の clusterBy 配列 ([]) は、ファイル圧縮のみを示します。

たとえば、 date 列と region 列によってデータをクラスター化する操作は、次のようになります。

operationParameters: {
  "clusterBy": "[\"date\",\"region\"]"
}

液体クラスタリングの詳細については、「テーブルに 液体クラスタリングを使用する」を参照してください。

Z順序

Z オーダーでは、 zOrderBy パラメーターに Z オーダー列名が設定されます。 空の zOrderBy 配列 ([]) は、操作で Z オーダーが適用されなかったことを示します。

たとえば、 date 列に Z オーダーを適用した操作は、次のようになります。

operationParameters: {
  "zOrderBy": "[\"date\"]"
}

操作スコープ

predicate パラメーターは、操作がテーブル全体で実行されたか、その一部でのみ実行されたかを示します。

• 空の predicate 配列 ([]) は、操作がテーブル全体で実行されたことを意味します。
• 設定された predicate 配列は、述語に一致するパーティションでのみ実行されたターゲット OPTIMIZE table_name WHERE <partition_predicate> コマンドを意味します。

たとえば、 year = 2024 に一致するパーティションを対象とする操作は、次のようになります。

operationParameters: {
  "predicate": "[\"'year = 2024\"]"
}

タイム トラベル

タイム トラベルでは、タイムスタンプまたはテーブルのバージョン (トランザクション ログに記録される) に基づいて、以前のテーブル バージョンのクエリがサポートされます。 タイム トラベルは、次のようなアプリケーションに使用できます。

• 機械学習モデルの出力など、分析、レポート、または出力を再作成する。 これは、特に規制対象の業界で、デバッグや監査に役立つ場合があります。
• 複雑なテンポラル クエリを記述する。
• データの間違いを修正する。
• 急速に変化するテーブルに対応する一連のクエリに対してスナップショット分離を提供する。

タイム トラベル構文

テーブル名の指定の後に句を追加して、タイム トラベルを使用してテーブルにクエリを実行します。

• timestamp_expression には次のいずれかを指定できます。
  • '2018-10-18T22:15:12.013Z'、つまり、タイムスタンプにキャストできる文字列です
  • cast('2018-10-18 13:36:32 CEST' as timestamp)
  • '2018-10-18'、つまり、日付文字列です
  • current_timestamp() - interval 12 hours
  • date_sub(current_date(), 1)
  • タイムスタンプにキャストされる (できる) その他の式
• version は、DESCRIBE HISTORY table_spec の出力から取得できる long 型の値です。

timestamp_expression も version もサブクエリにすることはできません。

日付またはタイムスタンプ文字列のみが使用できます。 たとえば、"2019-01-01" と"2019-01-01T00:00:00.000Z" です。 構文の例については、次のコードを参照してください。

SQL

SELECT * FROM people10m TIMESTAMP AS OF '2018-10-18T22:15:12.013Z';
SELECT * FROM people10m VERSION AS OF 123;

Python

df1 = spark.read.option("timestampAsOf", "2019-01-01").table("people10m")
df2 = spark.read.option("versionAsOf", 123).table("people10m")

また、@ 構文を使用して、タイムスタンプまたはバージョンをテーブル名の一部として指定することもできます。 タイムスタンプは yyyyMMddHHmmssSSS 形式である必要があります。 @vを使用してバージョンを指定できます。 構文の例については、次のコードを参照してください。

SQL

-- Timestamp version
SELECT * FROM people10m@20190101000000000
-- Version number
SELECT * FROM people10m@v123

Python

# Timestamp version
spark.read.table("people10m@20190101000000000")
# Version number
spark.read.table("people10m@v123")

タイム トラベル クエリのデータ保持を構成する

以前のバージョンのテーブルに対してクエリを実行するには、そのバージョンのログ ファイルとデータ ファイルの 両方 を保持する必要があります。

• データ ファイルは、テーブルに対して VACUUM を実行すると削除されます。
• ログ ファイルは、テーブルのバージョンをチェックポイント処理した後に自動的に削除されます。

テーブルのデータ保持しきい値を増やすには、 <format> を delta または icebergに置き換えて、次のテーブルのプロパティを構成する必要があります。

• <format>.logRetentionDuration = "interval <interval>": テーブルの履歴を保持する期間を制御します。 既定値は interval 30 days です。
  • Databricks Runtime 18.0 以降では、 logRetentionDuration は deletedFileRetentionDuration以上である必要があります。 Unity カタログのマネージド テーブルの場合、これは Databricks Runtime 12.2 以降に適用されます。
• <format>.deletedFileRetentionDuration = "interval <interval>": 現在のテーブル バージョンで参照されなくなったデータ ファイルを削除するために VACUUM で使用するしきい値を決定します。 既定値は interval 7 days です。

たとえば、30 日間の履歴データにアクセスするには、delta.deletedFileRetentionDuration = "interval 30 days"の既定の設定と一致するdelta.logRetentionDurationを設定します。

テーブルの作成時にテーブルのプロパティを指定することも、 ALTER TABLE ステートメントで設定することもできます。 テーブル のプロパティのリファレンスを参照してください。

タイム トラベルの例

ユーザー 111のテーブルへの誤った削除を修正するには:

INSERT INTO my_table
  SELECT * FROM my_table TIMESTAMP AS OF date_sub(current_date(), 1)
  WHERE userId = 111

テーブルに対する誤って行われた更新を修正するには:

MERGE INTO my_table target
  USING my_table TIMESTAMP AS OF date_sub(current_date(), 1) source
  ON source.userId = target.userId
  WHEN MATCHED THEN UPDATE SET *

過去 1 週間に追加された新しい顧客の数を照会するには:

SELECT
(
  SELECT count(distinct userId)
  FROM my_table
)
-
(
  SELECT count(distinct userId)
  FROM my_table TIMESTAMP AS OF date_sub(current_date(), 7)
) AS new_customers

トランザクション ログ チェックポイント

トランザクション ログは、テーブル のバージョンを、テーブル データと共にトランザクション ログ ディレクトリ内の JSON ファイルとして記録します。

チェックポイント クエリを最適化するために、テーブルのバージョンは Parquet チェックポイント ファイルに集計されるため、テーブル履歴のすべての JSON バージョンを読み取る必要がなくなり、パフォーマンスが向上します。 ユーザーはチェックポイントを直接操作する必要はありません。

Azure Databricks を使用すると、データ サイズとワークロードのチェックポイント処理の頻度が最適化されます。 チェックポイント処理の頻度は、予告なく変更される場合があります。

テーブルを以前の状態に復元する

RESTORE コマンドを使用して、次のシナリオを含め、テーブルを以前のバージョンまたはタイムスタンプに復元します。

• 既に復元されたテーブルを復元できます。
• 複製されたテーブルを復元できます。

次の要件を考慮してください。

• テーブルを復元するには、テーブルの MODIFY 権限が必要です。
• データ ファイルが削除された後、手動で、または VACUUMによって、それらのファイルを参照する古いバージョンにテーブルを復元することはできません。 それでも、spark.sql.files.ignoreMissingFiles が true に設定されている場合は、このバージョンに部分的に復元できます。
• タイムスタンプで復元するには、 yyyy-MM-dd HH:mm:ss 形式または yyyy-MM-dd形式を使用します。

RESTORE TABLE target_table TO VERSION AS OF <version>;
RESTORE TABLE target_table TO TIMESTAMP AS OF <timestamp>;

構文の詳細については、RESTOREを参照してください。

ストリーミング動作

復元はデータを変更する操作であり、ダウンストリーム ワークロードのデータが重複する可能性があります。 RESTORE コマンドによって追加されたログ エントリには、dataChange が true に設定されています。

テーブルへの更新を処理 する構造化ストリーミング ジョブなどのダウンストリーム ワークロードの場合、復元操作によって追加されたデータ変更ログ エントリは新しいデータ更新と見なされ、それらの処理によってデータが重複する可能性があります。

例えば次が挙げられます。

前の例では、 RESTORE コマンドを実行すると、テーブル バージョン 0 と 1 の読み取り時に以前に表示されていた更新が行われます。 ストリーミング クエリでこのテーブルが再度読み取られた場合、これらのファイルは新しく追加されたデータと見なされ、再度処理されます。

メトリックを復元

完了後、 RESTORE は次のメトリックを 1 行の DataFrame として報告します。

• table_size_after_restore: 復元後のテーブルのサイズ。

• num_of_files_after_restore: 復元後のテーブル内のファイルの数。

• num_removed_files: テーブルから削除された (論理的に削除された) ファイルの数。

• num_restored_files: ロールバックによって復元されたファイルの数。

• removed_files_size: テーブルから削除されたファイルの合計サイズ (バイト単位)。

• restored_files_size: 復元されたファイルの合計サイズ (バイト単位)。

  

最後のコミット バージョンを検索する

すべてのスレッドとすべてのテーブルの全体で、現在の SparkSession によって書き込まれた最後のコミットのバージョン番号を取得するには、SQL 構成 spark.databricks.<format>.lastCommitVersionInSession のクエリを実行します。 テーブルの形式に応じて、 <format> を delta または icebergに置き換えます。

例えば次が挙げられます。

SQL

SET spark.databricks.delta.lastCommitVersionInSession

Python

spark.conf.get("spark.databricks.delta.lastCommitVersionInSession")

Scala

spark.conf.get("spark.databricks.delta.lastCommitVersionInSession")

SparkSession によるコミットが行われていない場合は、キーのクエリを実行すると空の値が返されます。