テーブル履歴の操作

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

テーブル履歴の取得

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

テーブル履歴の保持期間は、テーブル設定 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 ドキュメントを参照してください。

カタログ エクスプローラー では、この詳細なテーブル情報と履歴が視覚的に表示されます。 テーブル スキーマとサンプル データに加えて、[履歴] タブをクリックして DESCRIBE HISTORY で表示されるテーブル履歴を確認することもできます。

履歴スキーマ

history 操作の出力には、次の列があります。

+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-----------------+-------------+--------------------+
|version|          timestamp|userId|userName|operation| operationParameters| job|notebook|clusterId|readVersion|   isolationLevel|isBlindAppend|    operationMetrics|
+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-----------------+-------------+--------------------+
|      5|2019-07-29 14:07:47|   ###|     ###|   DELETE|[predicate -> ["(...|null|     ###|      ###|          4|WriteSerializable|        false|[numTotalRows -> ...|
|      4|2019-07-29 14:07:41|   ###|     ###|   UPDATE|[predicate -> (id...|null|     ###|      ###|          3|WriteSerializable|        false|[numTotalRows -> ...|
|      3|2019-07-29 14:07:29|   ###|     ###|   DELETE|[predicate -> ["(...|null|     ###|      ###|          2|WriteSerializable|        false|[numTotalRows -> ...|
|      2|2019-07-29 14:06:56|   ###|     ###|   UPDATE|[predicate -> (id...|null|     ###|      ###|          1|WriteSerializable|        false|[numTotalRows -> ...|
|      1|2019-07-29 14:04:31|   ###|     ###|   DELETE|[predicate -> ["(...|null|     ###|      ###|          0|WriteSerializable|        false|[numTotalRows -> ...|
|      0|2019-07-29 14:01:40|   ###|     ###|    WRITE|[mode -> ErrorIfE...|null|     ###|      ###|       null|WriteSerializable|         true|[numFiles -> 2, n...|
+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-----------------+-------------+--------------------+

操作パラメーターの理解partitionBy

partitionBy フィールドは、テーブルのパーティション スキーマを定義または変更する CREATE 操作と OVERWRITE 操作でのみ意味があります。

既存のテーブルへの追加操作 (APPEND、INSERT、UPDATE、DELETE、MERGE) の場合、このフィールドには、使用される書き込み方法 ([]と.save()) に応じて、空の配列.saveAsTable()またはパーティション列が表示される場合があります。 この不整合は想定される動作であり、書き込みの検証には使用しないでください。

Example

date列でパーティション分割されたテーブルについて考えてみましょう。

# Initial table creation - partitionBy is populated
df.write.format("delta") \
  .partitionBy("date") \
  .saveAsTable("sales_data")

履歴の CREATE 操作には、次の情報が表示されます。

operationParameters: {
  "mode": "ErrorIfExists",
  "partitionBy": "[\"date\"]"
}

このテーブルにデータを追加する場合:

# Subsequent append - partitionBy shows empty
new_df.write.format("delta") \
  .mode("append") \
  .saveAsTable("sales_data")

APPEND 操作は次を示します。

operationParameters: {
  "mode": "Append",
  "partitionBy": "[]"
}

空の partitionBy 値が必要です。 データは、テーブルの既存のパーティション スキーマに基づいて正しいパーティションに書き込まれます。 パスに .save() すると、このフィールドにパーティション列が表示されることがありますが、この違いは実装の詳細であり、書き込み動作には影響しません。

操作メトリック

history操作は、operationMetrics 列マップ内の操作メトリックのコレクションを返します。

次の表は、マップのキー定義を操作別に示しています。

タイム トラベルとは

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

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

タイム トラベル構文

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

• 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

SELECT * FROM people10m@20190101000000000
SELECT * FROM people10m@v123

Python

spark.read.table("people10m@20190101000000000")
spark.read.table("people10m@v123")

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

テーブルのバージョンは、テーブル データと共に格納されるトランザクション ログ ディレクトリ内に JSON ファイルとして記録されます。 チェックポイント クエリを最適化するために、テーブルのバージョンは Parquet チェックポイント ファイルに集計されるため、テーブル履歴のすべての JSON バージョンを読み取る必要はありません。 Azure Databricks を使用すると、データ サイズとワークロードのチェックポイント処理の頻度が最適化されます。 ユーザーはチェックポイントを直接操作する必要はありません。 チェックポイント処理の頻度は、予告なく変更される場合があります。

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

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

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

ほとんどのテーブルはVACUUMが定期的に実行されるため、ポイントインタイムクエリはVACUUMの保持しきい値 (既定では7日間)を考慮する必要があります。

テーブルのデータ保持しきい値を増やすには、次のテーブル プロパティを構成する必要があります。

• delta.logRetentionDuration = "interval <interval>": テーブルの履歴を保持する期間を制御します。 既定では、 interval 30 daysです。
• delta.deletedFileRetentionDuration = "interval <interval>": 現在のテーブル バージョンで参照されなくなったデータ ファイルを削除するために VACUUM で使用するしきい値を決定します。 既定では、 interval 7 daysです。

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

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

RESTORE コマンドを使用して、テーブルを以前の状態に復元できます。 テーブルは、以前の状態に復元できる履歴バージョンを内部的に保持します。 RESTORE コマンドのオプションとして、以前の状態に対応するバージョンまたは以前の状態が作成されたときのタイムスタンプがサポートされています。

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

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

復元のメトリック

RESTORE 操作が完了すると、次のメトリックが単一行データフレームとして報告されます。

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

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

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

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

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

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

  

タイム トラベルの使用例

• ユーザー 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 *
  

• 先週に追加された新しい顧客の数を照会する:

  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
  

Spark セッションで最後のコミットのバージョンを見つける方法

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

SQL

SET spark.databricks.delta.lastCommitVersionInSession

Python

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

スカラ (プログラミング言語)

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

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