Bekerja dengan riwayat tabel Delta Lake

Setiap operasi yang memodifikasi tabel Delta Lake membuat versi tabel baru. Anda dapat menggunakan informasi riwayat untuk mengaudit operasi, memutar kembali tabel, atau mengkueri tabel pada titik waktu tertentu menggunakan perjalanan waktu.

Mengambil riwayat tabel Delta

Anda dapat mengambil informasi termasuk operasi, pengguna, dan tanda waktu untuk setiap penulisan ke tabel Delta dengan menjalankan history perintah . Operasi dikembalikan dalam urutan kronologis terbalik.

Retensi riwayat tabel ditentukan oleh pengaturan delta.logRetentionDurationtabel , yaitu 30 hari secara default.

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

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

Untuk detail sintaks Spark SQL, lihat DESCRIBE HISTORY.

Lihat dokumentasi Delta Lake API untuk detail sintaks Scala/Java/Python.

Catalog Explorer menyediakan tampilan visual informasi dan riwayat tabel terperinci ini untuk tabel Delta. Selain skema tabel dan data sampel, Anda dapat mengklik tab Riwayat untuk melihat riwayat tabel yang ditampilkan dengan DESCRIBE HISTORY.

Skema riwayat

Output operasi history memiliki kolom berikut.

+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-----------------+-------------+--------------------+
|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...|
+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-----------------+-------------+--------------------+

Kunci metrik operasi

Operasi history mengembalikan kumpulan metrik operasi di peta kolom operationMetrics.

Tabel berikut mencantumkan definisi kunci peta berdasarkan operasi.

Apa itu perjalanan waktu Delta Lake?

Perjalanan waktu Delta Lake mendukung kueri versi tabel sebelumnya berdasarkan tanda waktu atau versi tabel (seperti yang dicatat dalam log transaksi). Anda dapat menggunakan perjalanan waktu untuk aplikasi seperti berikut:

• Membuat ulang analisis, laporan, atau keluaran (misalnya, keluaran model pembelajaran mesin). Ini bisa berguna untuk debugging atau audit, terutama di industri yang diatur.
• Menulis pertanyaan temporal yang kompleks.
• Memperbaiki kesalahan dalam data Anda.
• Menyediakan isolasi snapshot untuk sekumpulan kueri untuk tabel yang berubah dengan cepat.

Sintaks perjalanan waktu delta

Anda mengkueri tabel Delta dengan perjalanan waktu dengan menambahkan klausa setelah spesifikasi nama tabel.

• timestamp_expression dapat berupa salah satu dari:
  • '2018-10-18T22:15:12.013Z', yaitu, string yang dapat ditransmisikan ke stempel waktu
  • cast('2018-10-18 13:36:32 CEST' as timestamp)
  • '2018-10-18', yaitu, string tanggal
  • current_timestamp() - interval 12 hours
  • date_sub(current_date(), 1)
  • Ekspresi lain yang sedang atau dapat ditransmisikan ke stempel waktu
• version adalah nilai panjang yang dapat diperoleh dari output DESCRIBE HISTORY table_spec.

Baik timestamp_expression maupun version tidak boleh berupa subkueri.

Hanya string tanggal atau tanda waktu yang diterima. Misalnya, "2019-01-01" dan "2019-01-01T00:00:00.000Z". Lihat kode berikut untuk contoh sintaks:

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")

Anda juga dapat menggunakan @ sintaks untuk menentukan tanda waktu atau versi sebagai bagian dari nama tabel. Stempel waktu harus dalam format yyyyMMddHHmmssSSS. Anda dapat menentukan versi setelah @ dengan menambahkan v ke versi tersebut. Lihat kode berikut untuk contoh sintaks:

SQL

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

Python

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

Apa itu titik pemeriksaan log transaksi?

Delta Lake merekam versi tabel sebagai file JSON dalam _delta_log direktori, yang disimpan bersama data tabel. Untuk mengoptimalkan kueri titik pemeriksaan, Delta Lake mengagregasi versi tabel ke file titik pemeriksaan Parquet, mencegah kebutuhan untuk membaca semua riwayat tabel versi JSON. Azure Databricks mengoptimalkan frekuensi titik pemeriksaan untuk ukuran data dan beban kerja. Pengguna tidak perlu berinteraksi dengan titik pemeriksaan secara langsung. Frekuensi titik pemeriksaan dapat berubah tanpa pemberitahuan.

Mengonfigurasi retensi data untuk kueri perjalanan waktu

Untuk mengkueri versi tabel sebelumnya, Anda harus menyimpan log dan file data untuk versi tersebut.

File data dihapus saat VACUUM dijalankan terhadap tabel. Delta Lake mengelola penghapusan file log secara otomatis setelah mengecek versi tabel.

Karena sebagian besar tabel Delta telah VACUUM berjalan terhadapnya secara teratur, kueri point-in-time harus menghormati ambang retensi untuk VACUUM, yaitu 7 hari secara default.

Untuk meningkatkan ambang retensi data untuk tabel Delta, Anda harus mengonfigurasi properti tabel berikut:

• delta.logRetentionDuration = "interval <interval>": mengontrol lama riwayat untuk tabel disimpan. Default adalah interval 30 days.
• delta.deletedFileRetentionDuration = "interval <interval>": menentukan ambang batas VACUUM yang digunakan untuk menghapus file data yang tidak lagi dirujuk dalam versi tabel saat ini. Default adalah interval 7 days.

Anda dapat menentukan properti Delta selama pembuatan tabel atau mengaturnya dengan ALTER TABLE pernyataan. Lihat Referensi properti tabel Delta.

Memulihkan tabel Delta ke kondisi sebelumnya

Anda dapat memulihkan tabel Delta ke kondisi sebelumnya dengan menggunakan perintah RESTORE. Tabel Delta secara internal mempertahankan versi historis tabel yang memungkinkannya dipulihkan ke kondisi sebelumnya. Versi yang sesuai dengan kondisi sebelumnya atau stempel waktu saat kondisi sebelumnya dibuat didukung sebagai opsi oleh perintah RESTORE.

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

Untuk detail sintaks, lihat RESTORE.

Pulihkan metrik

RESTORE melaporkan metrik berikut sebagai DataFrame baris tunggal setelah operasi selesai:

• table_size_after_restore: Ukuran tabel setelah memulihkan.

• num_of_files_after_restore: Jumlah file dalam tabel setelah memulihkan.

• num_removed_files: Jumlah file yang dihapus (dihapus secara logis) dari tabel.

• num_restored_files: Jumlah file yang dipulihkan karena bergulir kembali.

• removed_files_size: Ukuran total dalam byte file yang dihapus dari tabel.

• restored_files_size: Ukuran total dalam byte file yang dipulihkan.

  

Contoh penggunaan perjalanan waktu Delta Lake

• Memperbaiki penghapusan yang tidak disengaja ke tabel untuk pengguna 111:

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

• Memperbaiki pembaruan yang salah secara tidak sengaja ke tabel:

  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 *
  

• Tanyakan jumlah pelanggan baru yang ditambahkan selama seminggu terakhir.

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

Bagaimana cara menemukan versi penerapan terakhir dalam sesi Spark?

Untuk mendapatkan nomor versi dari penerapan terakhir yang ditulis oleh SparkSession saat ini di semua utas dan semua tabel, kueri konfigurasi SQL spark.databricks.delta.lastCommitVersionInSession.

SQL

SET spark.databricks.delta.lastCommitVersionInSession

Python

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

Scala

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

Jika tidak ada komit yang dibuat oleh SparkSession, mengkueri kunci mengembalikan nilai kosong.