Delta Lake table geçmişi üzerinde çalışma

Delta Lake table değiştiren her işlem yeni bir table sürümü oluşturur. Geçmiş bilgilerini kullanarak işlemleri denetleyebilir, tablegeri alabilir veya zaman yolculuğu kullanarak belirli bir noktada table sorgulayabilirsiniz.

Delta table geçmişini alma

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

Table geçmişi saklama, delta.logRetentionDurationtable ayarı ile belirlenir ve bu ayar varsayılan olarak 30 gündür.

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.

Catalog Explorer, Delta tablesiçin bu ayrıntılı table bilgilerinin ve geçmişinin görsel bir görünümünü sağlar. table schema ve örnek verilere ek olarak, DESCRIBE HISTORYile görüntülenen table geçmişini görmek için Geçmişi sekmesine tıklayabilirsiniz.

Geçmiş schema

history işleminin çıktısı aşağıdaki columns'e 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ı

history işlemi, operationMetricscolumn eşlemesinde bir işlem ölçümleri koleksiyonu döndürür.

Aşağıdaki tableslist harita anahtarı tanımları işlemlere göre verilmiştir.

Delta Lake zaman yolculuğu nedir?

Delta Lake zaman yolculuğu, zaman damgasına veya table sürümüne göre (işlem günlüğüne kaydedildiği gibi) önceki table 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 tablesiçin bir sorgu kümesi set'a yönelik anlık görüntü izolasyonu sağlama.

Delta zaman atlatma söz dizimi

table ad belirtiminin arkasına bir yan tümce ekleyerek zaman yolculuğu ile Delta table'ı sorgulayabilirsiniz.

• 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ü table 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, table sürümleri table veriyle birlikte depolanan _delta_log dizininde JSON dosyaları olarak kaydeder. optimize denetim noktası sorgulaması için, Delta Lake, table sürümlerini Parquet denetim noktası dosyalarına toplayarak table geçmişinin tüm JSON sürümlerini okuma gereğini ortadan kaldırır. 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 sürümünü sorgulamak için hem günlük hem de bu sürüme ilişkin veri dosyalarını tutmanız gerekir.

VACUUM bir tableüzerinde çalıştığında veri dosyaları silinir. Delta Lake, denetim noktası table sürümlerine geçtikten sonra günlük dosyasının kaldırılmasını otomatik olarak yönetir.

Çoğu Delta tables üzerinde düzenli olarak VACUUM çalıştırıldığından, anlık sorgular varsayılan olarak 7 gün olan VACUUMiçin bekletme eşiğine uygun olmalıdır.

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

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

Delta özelliklerini table oluşturma sırasında belirtebilir veya bir ALTER TABLE ifadesi ile set edebilirsiniz. Bkz. Delta table özellik referansı.

Restore Delta table'yi önceki bir duruma

restore komutunu kullanarak delta table'i önceki durumuna RESTORE yapabilirsiniz. Delta table, table'in önceki bir duruma geri yüklenmesini sağlayan geçmiş sürümlerini dahili olarak saklar. Ö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.

Restore ölçümleri

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

• table_size_after_restore: Geri yükleme sonrası table boyutu.

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

• num_removed_files: tablekaldı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: tablekaldı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

• Yanlışlıkla silinen table'ları kullanıcı 111için düzeltin.

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

• Bir table'da yanlışlıkla yapılan hatalı güncellemeyi düzeltin.

  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 tablesyazılan son işlemenin sürüm numarasını get 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.