如何進(jìn)行數(shù)據(jù)湖deltalake中的時(shí)間旅行及版本管理

本篇文章給大家分享的是有關(guān)如何進(jìn)行數(shù)據(jù)湖deltalake中的時(shí)間旅行及版本管理，小編覺得挺實(shí)用的，因此分享給大家學(xué)習(xí)，希望大家閱讀完這篇文章后可以有所收獲，話不多說，跟著小編一起來看看吧。

deltalake支持?jǐn)?shù)據(jù)版本管理和時(shí)間旅行：提供了數(shù)據(jù)快照，使開發(fā)人員能夠訪問和還原早期版本的數(shù)據(jù)以進(jìn)行審核、回滾或重新計(jì)算。

1.場景

delta lake的時(shí)間旅行，實(shí)際上就是利用多版本管理機(jī)制，查詢歷史的delta 表快照。時(shí)間旅行有以下使用案例：

1).可以重復(fù)創(chuàng)建數(shù)據(jù)分析，報(bào)告或者一些輸出(比如，機(jī)器學(xué)習(xí)模型)。這主要是有利于調(diào)試和安全審查，尤其是在受管制的行業(yè)里。

2).編寫復(fù)雜的基于時(shí)間的查詢。

3).修正數(shù)據(jù)中的錯(cuò)誤信息。

4).為一組查詢提供快照隔離，以快速變更表。

2.配置

DataframeTable支持創(chuàng)建dataframe的時(shí)候指定一個(gè)delta lake表的版本信息：

val df1 = spark.read.format("delta").option("timestampAsOf", timestamp_string).load("/delta/events")val df2 = spark.read.format("delta").option("versionAsOf", version).load("/delta/events")

對(duì)于版本號(hào)，直接傳入一個(gè)版本數(shù)值即可，如下：

val df2 = spark.read.format("delta").option("versionAsOf", 0).table(tableName)

對(duì)于timestamp字符串，必須要是date格式或者timestamp格式。例如：

val df1 = spark.read.format("delta").option("timestampAsOf", "2020-06-28").load("/delta/events")val df1 = spark.read.format("delta").option("timestampAsOf", "2020-06-28T00:00:00.000Z").load("/delta/events")

由于delta lake的表是存在更新的情況，所以多次讀取數(shù)據(jù)生成的dataframe之間會(huì)有差異，因?yàn)閮纱巫x取數(shù)據(jù)可能是一次是數(shù)據(jù)更新前，另一次是數(shù)據(jù)更新后。使用時(shí)間旅行你就可以在多次調(diào)用之間修復(fù)數(shù)據(jù)。

val latest_version = spark.sql("SELECT max(version) FROM (DESCRIBE HISTORY delta.`/delta/events`)").collect()val df = spark.read.format("delta").option("versionAsOf", latest_version[0][0]).load("/delta/events")

3.數(shù)據(jù)保存時(shí)間

默認(rèn)情況下，deltalake保存最近30天的提交歷史。這就意味著可以指定30天之前的版本來讀取數(shù)據(jù)，但是有些注意事項(xiàng)：

3.1 沒對(duì)delta 表調(diào)用VACUUM函數(shù)。VACUUM函數(shù)是用來刪除不在引用的delta表和一些超過保留時(shí)間的表，支持sql和API形式。

slq表達(dá)式：

VACUUM eventsTable   -- vacuum files not required by versions older than the default retention period
VACUUM '/data/events' -- vacuum files in path-based table
VACUUM delta.`/data/events/`
VACUUM delta.`/data/events/` RETAIN 100 HOURS  -- vacuum files not required by versions more than 100 hours old
VACUUM eventsTable DRY RUN    -- do dry run to get the list of files to be deleted

?

scala API 表達(dá)式

import io.delta.tables._
val deltaTable = DeltaTable.forPath(spark, pathToTable)
deltaTable.vacuum()        // vacuum files not required by versions older than the default retention period
deltaTable.vacuum(100)     // vacuum files not required by versions more than 100 hours old

可以通過下面兩個(gè)delta 表屬性配置來

• delta.logRetentionDuration =“ interval <interval>”：控制將表的歷史記錄保留多長時(shí)間。每次寫入checkpoint時(shí)，都會(huì)自動(dòng)清除早于保留間隔的日志。如果將此配置設(shè)置為足夠大的值，則會(huì)保留許多日志。這不會(huì)影響性能，因?yàn)獒槍?duì)日志的操作是常量時(shí)間。歷史記錄的操作是并行的（但是隨著日志大小的增加，它將變得更加耗時(shí)）。默認(rèn)值為 interval 30 days。

• delta.deletedFileRetentionDuration =“ interval <interval>”：在這個(gè)時(shí)間范圍內(nèi)的數(shù)據(jù)是不會(huì)被VACUUM命令刪除。默認(rèn)值為間隔7天。要訪問30天的歷史數(shù)據(jù)，請(qǐng)?jiān)O(shè)置delta.deletedFileRetentionDuration = "interval 30 days"。此設(shè)置可能會(huì)導(dǎo)致您的存儲(chǔ)成本上升。

注意：VACUUM命令是不會(huì)刪除日志文件的，日志文件是在checkpoint之后自動(dòng)刪除的。

為了讀取之前版本的數(shù)據(jù)，必須要保留該版本的日志文件和數(shù)據(jù)文件。

4.案例

修復(fù)意外刪除的用戶111的數(shù)據(jù)。

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

修復(fù)錯(cuò)誤更新的數(shù)據(jù)

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 *

查詢過去七天新增的消費(fèi)者數(shù)：

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

以上就是如何進(jìn)行數(shù)據(jù)湖deltalake中的時(shí)間旅行及版本管理，小編相信有部分知識(shí)點(diǎn)可能是我們?nèi)粘９ぷ鲿?huì)見到或用到的。希望你能通過這篇文章學(xué)到更多知識(shí)。更多詳情敬請(qǐng)關(guān)注億速云行業(yè)資訊頻道。