Delta Lake tablo geçmişiyle çalışma

Delta Lake tablosunu değiştiren her işlem yeni bir tablo sürümü oluşturur. Zaman yolculuğu kullanarak işlemleri denetlemek, bir tabloyu geri almak veya belirli bir zamanda tabloyu sorgulamak için geçmiş bilgilerini kullanabilirsiniz.

Delta tablosu geçmişini alma

Komutunu çalıştırarak history Delta tablosuna yapılan her yazma işlemi, kullanıcı ve zaman damgası gibi bilgileri alabilirsiniz. İşlemler ters kronolojik sırayla döndürülür.

Tablo geçmişi saklama, varsayılan olarak 30 gün olan tablo ayarı delta.logRetentionDurationtarafından belirlenir.

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

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

Spark SQL söz dizimi ayrıntıları için bkz . DESCRIBE HISTORY.

Scala/Java/Python söz dizimi ayrıntıları için Delta Lake API belgelerine bakın.

Katalog Gezgini , Delta tabloları için bu ayrıntılı tablo bilgilerinin ve geçmişinin görsel bir görünümünü sağlar. Tablo şemasına ve örnek verilere ek olarak, ile DESCRIBE HISTORYgörüntülenen tablo geçmişini görmek için Geçmiş sekmesine tıklayabilirsiniz.

Geçmiş şeması

İşlemin çıktısı history aşağıdaki sütunlara sahiptir.

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

İşlem ölçümleri anahtarları

İşlem, history sütun haritasında operationMetrics bir işlem ölçümleri koleksiyonu döndürür.

Aşağıdaki tablolarda eşleme anahtarı tanımları işleme göre listelemektedir.

Delta Lake zaman yolculuğu nedir?

Delta Lake zaman yolculuğu, zaman damgasına veya tablo sürümüne göre (işlem günlüğünde kaydedildiği gibi) önceki tablo sürümlerini sorgulamayı destekler. Aşağıdaki gibi uygulamalar için zaman yolculuğu kullanabilirsiniz:

• Çözümlemeleri, raporları veya çıkışları yeniden oluşturma (örneğin, bir makine öğrenmesi modelinin çıktısı). Bu, özellikle düzenlemeye tabi sektörlerde hata ayıklama veya denetim için yararlı olabilir.
• Karmaşık zamana bağlı sorgular yazma.
• Verilerinizdeki hataları düzeltme.
• Hızlı değişen tablolara yönelik bir dizi sorgu için anlık görüntü yalıtımı sağlama.

Delta zaman atlatma söz dizimi

Tablo adı belirtiminin arkasına bir yan tümce ekleyerek bir Delta tablosunu zaman yolculuğuyla sorgularsınız.

• timestamp_expression şu türlerden herhangi biri olabilir:
  • '2018-10-18T22:15:12.013Z', başka bir ifadeyle zaman damgasına atanabilen bir dizedir
  • cast('2018-10-18 13:36:32 CEST' as timestamp)
  • '2018-10-18', yani bir tarih dizesi
  • current_timestamp() - interval 12 hours
  • date_sub(current_date(), 1)
  • Zaman damgasına atanabilen veya atanabilen diğer tüm ifadeler
• version , çıkışından elde edilebilen uzun bir değerdir DESCRIBE HISTORY table_spec.

Alt timestamp_expression sorgular da version olamaz.

Yalnızca tarih veya zaman damgası dizeleri kabul edilir. Örneğin, "2019-01-01" ve "2019-01-01T00:00:00.000Z". Söz dizimi gibi aşağıdaki koda bakın:

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

Zaman damgasını @ veya sürümü tablo adının bir parçası olarak belirtmek için söz dizimini de kullanabilirsiniz. Zaman damgası biçiminde yyyyMMddHHmmssSSS olmalıdır. sürümünden sonra @ bir ön koşula bağlanarak bir v sürüm belirtebilirsiniz. Söz dizimi gibi aşağıdaki koda bakın:

SQL

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

Python

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

İşlem günlüğü denetim noktaları nedir?

Delta Lake, tablo sürümlerini dizin içinde _delta_log JSON dosyaları olarak kaydeder ve bu dosyalar tablo verileriyle birlikte depolanır. Kontrol noktası sorgulamasını iyileştirmek için Delta Lake tablo sürümlerini Parquet denetim noktası dosyalarına toplayarak tablo geçmişinin tüm JSON sürümlerini okuma gereksinimini önler. Azure Databricks, veri boyutu ve iş yükü için denetim noktası oluşturma sıklığını iyileştirir. Kullanıcıların denetim noktalarıyla doğrudan etkileşim kurması gerekmez. Denetim noktası sıklığı bildirimde bulunmadan değiştirilebilir.

Zaman yolculuğu sorguları için veri saklamayı yapılandırma

Önceki bir tablo sürümünü sorgulamak için, bu sürüm için hem günlüğü hem de veri dosyalarını tutmanız gerekir.

Bir tabloda çalıştırıldığında VACUUM veri dosyaları silinir. Delta Lake, tablo sürümlerini denetledikten sonra günlük dosyasının kaldırılmasını otomatik olarak yönetir.

Delta tablolarının VACUUM çoğu bunlara karşı düzenli olarak çalıştığından, belirli bir noktaya sorgular için VACUUMvarsayılan olarak 7 gün olan bekletme eşiğine uygun olmalıdır.

Delta tablolarında veri saklama eşiğini artırmak için aşağıdaki tablo özelliklerini yapılandırmanız gerekir:

• delta.logRetentionDuration = "interval <interval>": bir tablonun geçmişinin ne kadar süreyle tutulduğunu denetler. Varsayılan değer: interval 30 days.
• delta.deletedFileRetentionDuration = "interval <interval>": geçerli tablo sürümünde artık başvurulmayan veri dosyalarını kaldırmak için kullanılan eşiği VACUUM belirler. Varsayılan değer: interval 7 days.

Tablo oluşturma sırasında Delta özelliklerini belirtebilir veya bunları bir ALTER TABLE deyimiyle ayarlayabilirsiniz. Bkz . Delta tablosu özellikleri başvurusu.

Delta tablosunu önceki bir duruma geri yükleme

Komutunu kullanarak RESTORE Delta tablosunu önceki durumuna geri yükleyebilirsiniz. Delta tablosu, tablonun önceki bir duruma geri yüklenmesini sağlayan geçmiş sürümlerini dahili olarak korur. Önceki duruma veya önceki durumun oluşturulduğu zaman damgasına karşılık gelen bir sürüm, RESTORE komutu tarafından seçenekler olarak desteklenir.

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

Söz dizimi ayrıntıları için bkz . RESTORE.

Ölçümleri geri yükleme

RESTORE işlem tamamlandıktan sonra aşağıdaki ölçümleri tek satırlık DataFrame olarak bildirir:

• table_size_after_restore: Geri yüklemeden sonra tablonun boyutu.

• num_of_files_after_restore: Geri yüklemeden sonra tablodaki dosya sayısı.

• num_removed_files: Tablodan kaldırılan (mantıksal olarak silinen) dosya sayısı.

• num_restored_files: Geri dönme nedeniyle geri yüklenen dosyaların sayısı.

• removed_files_size: Tablodan kaldırılan dosyaların bayt cinsinden toplam boyutu.

• restored_files_size: Geri yüklenen dosyaların bayt cinsinden toplam boyutu.

  

Delta Lake zaman atlatma kullanımına örnekler

• Kullanıcının 111tabloya yanlışlıkla silmelerini düzeltin:

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

• Tabloya yanlışlıkla yapılan yanlış güncelleştirmeler düzeltilir:

  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 *
  

• Geçen hafta eklenen yeni müşterilerin sayısını sorgula.

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

Spark oturumunda son işlemenin sürümünü Nasıl yaparım? buldunuz?

Geçerli SparkSession tarafından tüm iş parçacıklarında ve tüm tablolarda yazılan son işlemenin sürüm numarasını almak için SQL yapılandırmasını spark.databricks.delta.lastCommitVersionInSessionsorgular.

SQL

SET spark.databricks.delta.lastCommitVersionInSession

Python

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

Scala

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

tarafından SparkSessionhiçbir işleme yapılmadıysa, anahtarın sorgulanması boş bir değer döndürür.