Översikt över Linux Foundation Delta Lake

Artikel
08/15/2023

Den här artikeln har anpassats för tydlighetens skull från den ursprungliga motsvarigheten här. Den här artikeln hjälper dig att snabbt utforska de viktigaste funktionerna i Delta Lake. Artikeln innehåller kodfragment som visar hur du läser från och skriver till Delta Lake-tabeller från interaktiva frågor, batchfrågor och strömningsfrågor. Kodfragmenten är också tillgängliga i en uppsättning notebook-filer PySpark här, Scala här och C# här

Här är vad vi kommer att täcka:

Skapa en tabell
Läsa data
Uppdatera tabelldata
Skriv över tabelldata
Villkorlig uppdatering utan överskrivning
Läsa äldre versioner av data med time travel
Skriva en dataström till en tabell
Läsa en dataström med ändringar från en tabell
SQL-stöd

Konfiguration

Se till att du ändrar nedanstående efter behov för din miljö.

import random

session_id = random.randint(0,1000000)
delta_table_path = "/delta/delta-table-{0}".format(session_id)

delta_table_path

var sessionId = (new Random()).Next(10000000);
var deltaTablePath = $"/delta/delta-table-{sessionId}";

deltaTablePath

val sessionId = scala.util.Random.nextInt(1000000)
val deltaTablePath = s"/delta/delta-table-$sessionId";

Resulterar i:

'/delta/delta-table-335323'

Skapa en tabell

Om du vill skapa en Delta Lake-tabell skriver du en DataFrame ut en DataFrame i deltaformat. Du kan ändra formatet från Parquet, CSV, JSON och så vidare till delta.

Koden nedan visar hur du skapar en ny Delta Lake-tabell med hjälp av schemat som härleds från din DataFrame.

data = spark.range(0,5)
data.show()
data.write.format("delta").save(delta_table_path)

var data = spark.Range(0,5);
data.Show();
data.Write().Format("delta").Save(deltaTablePath);

val data = spark.range(0, 5)
data.show
data.write.format("delta").save(deltaTablePath)

Resulterar i:

ID
0
1
2
3
4

Läsa data

Du läser data i Delta Lake-tabellen genom att ange sökvägen till filerna och deltaformatet.

df = spark.read.format("delta").load(delta_table_path)
df.show()

var df = spark.Read().Format("delta").Load(deltaTablePath);
df.Show()

val df = spark.read.format("delta").load(deltaTablePath)
df.show()

Resulterar i:

ID
1
3
4
0
2

Resultatordningen skiljer sig från ovan eftersom ingen ordning uttryckligen angavs innan resultatet matas ut.

Uppdatera tabelldata

Delta Lake stöder flera åtgärder för att ändra tabeller med hjälp av standard-API:er för DataFrame. Dessa åtgärder är en av förbättringarna som deltaformatet lägger till. I följande exempel körs ett batchjobb för att skriva över data i tabellen.

data = spark.range(5,10)
data.write.format("delta").mode("overwrite").save(delta_table_path)
df.show()

var data = spark.Range(5,10);
data.Write().Format("delta").Mode("overwrite").Save(deltaTablePath);
df.Show();

val data = spark.range(5, 10)
data.write.format("delta").mode("overwrite").save(deltaTablePath)
df.show()

Resulterar i:

ID
7
8
5
9
6

Här kan du se att alla fem posterna har uppdaterats för att innehålla nya värden.

Spara som katalogtabeller

Delta Lake kan skriva till hanterade eller externa katalogtabeller.

data.write.format("delta").saveAsTable("ManagedDeltaTable")
spark.sql("CREATE TABLE ExternalDeltaTable USING DELTA LOCATION '{0}'".format(delta_table_path))
spark.sql("SHOW TABLES").show()

data.Write().Format("delta").SaveAsTable("ManagedDeltaTable");
spark.Sql($"CREATE TABLE ExternalDeltaTable USING DELTA LOCATION '{deltaTablePath}'");
spark.Sql("SHOW TABLES").Show();

data.write.format("delta").saveAsTable("ManagedDeltaTable")
spark.sql(s"CREATE TABLE ExternalDeltaTable USING DELTA LOCATION '$deltaTablePath'")
spark.sql("SHOW TABLES").show

Resulterar i:

databas	tableName	isTemporary
standard	externaldeltatable	falskt
standard	manageddeltatable	falskt

Med den här koden skapade du en ny tabell i katalogen från en befintlig dataram, som kallas en hanterad tabell. Sedan definierade du en ny extern tabell i katalogen som använder en befintlig plats, som kallas för en extern tabell. I utdata visas båda tabellerna, oavsett hur de skapades, i katalogen.

Nu kan du titta på de utökade egenskaperna för båda dessa tabeller

spark.sql("DESCRIBE EXTENDED ManagedDeltaTable").show(truncate=False)

spark.Sql("DESCRIBE EXTENDED ManagedDeltaTable").Show(truncate: 0);

spark.sql("DESCRIBE EXTENDED ManagedDeltaTable").show(truncate=false)

Resulterar i:

col_name	data_type	kommentar
id	bigint	null

Detaljerad tabellinformation
Databas	standard
Tabell	manageddeltatable
Ägare	trusted-service-user
Genereringstid	lör 25 apr 00:35:34 UTC 2020
Senaste åtkomst	Tor jan 01 00:00:00 UTC 1970
Skapad av	Spark 2.4.4.2.6.99.201-11401300
Typ	HANTERADE
Leverantör	delta
Tabellegenskaper	[transient_lastDdlTime=1587774934]
Statistik	2 407 byte
Location	abfss://data@<data lake.dfs.core.windows.net/synapse/workspaces/>< arbetsytans namn>/lager/manageddeltatable
Serde-bibliotek	org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
InputFormat	org.apache.hadoop.mapred.SequenceFileInputFormat
OutputFormat	org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat
Lagringsegenskaper	[serialization.format=1]

spark.sql("DESCRIBE EXTENDED ExternalDeltaTable").show(truncate=False)

spark.Sql("DESCRIBE EXTENDED ExternalDeltaTable").Show(truncate: 0);

spark.sql("DESCRIBE EXTENDED ExternalDeltaTable").show(truncate=false)

Resultat i:

col_name	data_type	kommentar
id	bigint	null

Detaljerad tabellinformation
Databas	standard
Tabell	externaldeltatable
Ägare	trusted-service-user
Genereringstid	Lör 25 apr 00:35:38 UTC 2020
Senaste åtkomst	Tor jan 01 00:00:00 UTC 1970
Skapad av	Spark 2.4.4.2.6.99.201-11401300
Typ	EXTERNA
Leverantör	DELTA
Tabellegenskaper	[transient_lastDdlTime=1587774938]
Location	abfss://data@<data lake.dfs.core.windows.net/delta/delta-table-587152>
Serde-bibliotek	org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
InputFormat	org.apache.hadoop.mapred.SequenceFileInputFormat
OutputFormat	org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat
Lagringsegenskaper	[serialization.format=1]

Villkorsstyrd uppdatering utan att skriva över

Delta Lake tillhandahåller programmatiska API:er för villkorlig uppdatering, borttagning och sammanslagning (det här kommandot kallas ofta för en upsert) data i tabeller.

from delta.tables import *
from pyspark.sql.functions import *

delta_table = DeltaTable.forPath(spark, delta_table_path)

delta_table.update(
  condition = expr("id % 2 == 0"),
  set = { "id": expr("id + 100") })
delta_table.toDF().show()

using Microsoft.Spark.Extensions.Delta;
using Microsoft.Spark.Extensions.Delta.Tables;
using Microsoft.Spark.Sql;
using static Microsoft.Spark.Sql.Functions;

var deltaTable = DeltaTable.ForPath(deltaTablePath);

deltaTable.Update(
  condition: Expr("id % 2 == 0"),
  set: new Dictionary<string, Column>(){{ "id", Expr("id + 100") }});
deltaTable.ToDF().Show();

import io.delta.tables._
import org.apache.spark.sql.functions._

val deltaTable = DeltaTable.forPath(deltaTablePath)

// Update every even value by adding 100 to it
deltaTable.update(
  condition = expr("id % 2 == 0"),
  set = Map("id" -> expr("id + 100")))
deltaTable.toDF.show

Resultat i:

ID
106
108
5
7
9

Här har du precis lagt till 100 till varje jämnt ID.

delta_table.delete("id % 2 == 0")
delta_table.toDF().show()

deltaTable.Delete(condition: Expr("id % 2 == 0"));
deltaTable.ToDF().Show();

deltaTable.delete(condition = expr("id % 2 == 0"))
deltaTable.toDF.show

Resultat i:

ID
5
7
9

Observera att varje jämn rad har tagits bort.

new_data = spark.range(0,20).alias("newData")

delta_table.alias("oldData")\
    .merge(new_data.alias("newData"), "oldData.id = newData.id")\
    .whenMatchedUpdate(set = { "id": lit("-1")})\
    .whenNotMatchedInsert(values = { "id": col("newData.id") })\
    .execute()

delta_table.toDF().show(100)

var newData = spark.Range(20).As("newData");

deltaTable
    .As("oldData")
    .Merge(newData, "oldData.id = newData.id")
    .WhenMatched()
        .Update(new Dictionary<string, Column>() {{"id", Lit("-1")}})
    .WhenNotMatched()
        .Insert(new Dictionary<string, Column>() {{"id", Col("newData.id")}})
    .Execute();

deltaTable.ToDF().Show(100);

val newData = spark.range(0, 20).toDF

deltaTable.as("oldData").
  merge(
    newData.as("newData"),
    "oldData.id = newData.id").
  whenMatched.
  update(Map("id" -> lit(-1))).
  whenNotMatched.
  insert(Map("id" -> col("newData.id"))).
  execute()

deltaTable.toDF.show()

Resultat i:

ID
18
15
19
2
1
6
8
3
-1
10
13
0
16
4
-1
12
11
14
-1
17

Här har du en kombination av befintliga data. Befintliga data har tilldelats värdet -1 i kodsökvägen update(WhenMatched). De nya data som skapades överst i kodfragmentet och lades till via sökvägen infoga kod (WhenNotMatched) lades också till.

Historik

Delta Lake's har möjlighet att göra det möjligt att titta på historiken för en tabell. Det vill: de ändringar som har gjorts i den underliggande deltatabellen. Cellen nedan visar hur enkelt det är att inspektera historiken.

delta_table.history().show(20, 1000, False)

deltaTable.History().Show(20, 1000, false);

deltaTable.history.show(false)

Resultat i:

version	timestamp	userId	userName	operation	operationParameters	jobb	notebook-fil	clusterId	readVersion	isolationLevel	isBlindAppend
4	2020-04-25 00:36:27	null	null	SAMMANFOGA	[predikat -> (oldData.`ID` = newData.`ID`)]	null	null	null	3	null	falskt
3	2020-04-25 00:36:08	null	null	DELETE	[predikat -> ["((`ID` % CAST(2 AS BIGINT)) = CAST(0 AS BIGINT))"]]	null	null	null	2	null	falskt
2	2020-04-25 00:35:51	null	null	UPDATE	[predikat -> ((ID#744L % cast(2 as bigint)) = cast(0 as bigint))]	null	null	null	1	null	falskt
1	2020-04-25 00:35:05	null	null	SKRIVA	[mode -> Skriv över, partitionBy –> []]	null	null	null	0	null	falskt
0	2020-04-25 00:34:34	null	null	SKRIVA	[mode -> ErrorIfExists, partitionBy -> []]	null	null	null	null	null	true

Här kan du se alla ändringar som gjorts i kodfragmenten ovan.

Läsa äldre versioner av data med time travel

Du kan köra frågor mot tidigare ögonblicksbilder av Din Delta Lake-tabell med hjälp av en funktion som kallas Tidsresor. Om du vill komma åt de data som du skrev över kan du fråga en ögonblicksbild av tabellen innan du skriver över den första datauppsättningen med hjälp av alternativet versionAsOf.

När du har kört cellen nedan bör du se den första uppsättningen data från innan du skriver över den. Time Travel är en kraftfull funktion som utnyttjar kraften i Delta Lake-transaktionsloggen för att komma åt data som inte längre finns i tabellen. Om du tar bort alternativet version 0 (eller anger version 1) kan du se nyare data igen. Mer information finns i Fråga en äldre ögonblicksbild av en tabell.

df = spark.read.format("delta").option("versionAsOf", 0).load(delta_table_path)
df.show()

var df = spark.Read().Format("delta").Option("versionAsOf", 0).Load(deltaTablePath);
df.Show();

val df = spark.read.format("delta").option("versionAsOf", 0).load(deltaTablePath)
df.show()

Resulterar i:

ID
0
1
4
3
2

Här kan du se att du har gått tillbaka till den tidigaste versionen av data.

Skriva en dataström till en tabell

Du kan också skriva till en Delta Lake-tabell med hjälp av Sparks Structured Streaming. Delta Lake-transaktionsloggen garanterar bearbetning exakt en gång, även om det finns andra strömmar eller batchfrågor som körs samtidigt mot tabellen. Som standard körs strömmar i tilläggsläge, vilket lägger till nya poster i tabellen.

Mer information om Delta Lake-integrering med Structured Streaming finns i Läsa och skriva för tabellströmning.

I cellerna nedan gör vi följande:

Cell 30 Visa nyligen tillagda data
Cell 31 Inspektera historik
Cell 32 Stoppa det strukturerade direktuppspelningsjobbet
Cell 33 Inspektera historik <– Du kommer att märka att tilläggen har stoppats

Först ska du konfigurera ett enkelt Spark Streaming-jobb för att generera en sekvens och göra jobbet skrivet till deltatabellen.

streaming_df = spark.readStream.format("rate").load()
stream = streaming_df\
    .selectExpr("value as id")\
    .writeStream\
    .format("delta")\
    .option("checkpointLocation", "/tmp/checkpoint-{0}".format(session_id))\
    .start(delta_table_path)

var streamingDf = spark.ReadStream().Format("rate").Load();
var stream = streamingDf.SelectExpr("value as id").WriteStream().Format("delta").Option("checkpointLocation", $"/tmp/checkpoint-{sessionId}").Start(deltaTablePath);

val streamingDf = spark.readStream.format("rate").load()
val stream = streamingDf.select($"value" as "id").writeStream.format("delta").option("checkpointLocation", s"/tmp/checkpoint-$sessionId").start(deltaTablePath)

Läsa en dataström med ändringar från en tabell

När strömmen skrivs till Delta Lake-tabellen kan du också läsa från den tabellen som en strömningskälla. Du kan till exempel starta en annan strömmande fråga som skriver ut alla ändringar som gjorts i Delta Lake-tabellen.

delta_table.toDF().sort(col("id").desc()).show(100)

deltaTable.ToDF().Sort(Col("id").Desc()).Show(100);

deltaTable.toDF.sort($"id".desc).show

Resulterar i:

ID
19
18
17
16
15
14
13
12
11
10
8
6
4
3
2
1
0
-1
-1
-1

delta_table.history().drop("userId", "userName", "job", "notebook", "clusterId", "isolationLevel", "isBlindAppend").show(20, 1000, False)

deltaTable.History().Drop("userId", "userName", "job", "notebook", "clusterId", "isolationLevel", "isBlindAppend").Show(20, 1000, false);

deltaTable.history.show

Resulterar i:

version	timestamp	operation	operationParameters	readVersion
5	2020-04-25 00:37:09	DIREKTUPPSPELNINGSUPPDATERING	[outputMode –> Tillägg, queryId –> d26b4f8a-7e5a-44f2-a5fb-23a7bd02aef7, epochId -> 0]	4
4	2020-04-25 00:36:27	SAMMANFOGA	[predikat -> (oldData.`id` = newData.`id`)]	3
3	2020-04-25 00:36:08	DELETE	[predikat -> ["((`id` % CAST(2 AS BIGINT)) = CAST(0 AS BIGINT))"]]	2
2	2020-04-25 00:35:51	UPDATE	[predikat -> ((id#744L % cast(2 as bigint)) = cast(0 as bigint))]	1
1	2020-04-25 00:35:05	SKRIVA	[mode -> Skriv över, partitionBy –> []]	0
0	2020-04-25 00:34:34	SKRIVA	[mode -> ErrorIfExists, partitionBy -> []]	null

Här släpper du några av de mindre intressanta kolumnerna för att förenkla visningsupplevelsen i historikvyn.

stream.stop()
delta_table.history().drop("userId", "userName", "job", "notebook", "clusterId", "isolationLevel", "isBlindAppend").show(100, 1000, False)

stream.Stop();
deltaTable.History().Drop("userId", "userName", "job", "notebook", "clusterId", "isolationLevel", "isBlindAppend").Show(100, 1000, false);

stream.stop
deltaTable.history.show

Resulterar i:

version	timestamp	operation	operationParameters	readVersion
5	2020-04-25 00:37:09	DIREKTUPPSPELNINGSUPPDATERING	[outputMode –> Tillägg, queryId –> d26b4f8a-7e5a-44f2-a5fb-23a7bd02aef7, epochId -> 0]	4
4	2020-04-25 00:36:27	SAMMANFOGA	[predikat -> (oldData.`id` = newData.`id`)]	3
3	2020-04-25 00:36:08	DELETE	[predikat -> ["((`id` % CAST(2 AS BIGINT)) = CAST(0 AS BIGINT))"]]	2
2	2020-04-25 00:35:51	UPDATE	[predikat -> ((id#744L % cast(2 as bigint)) = cast(0 as bigint))]	1
1	2020-04-25 00:35:05	SKRIVA	[mode -> Skriv över, partitionBy –> []]	0
0	2020-04-25 00:34:34	SKRIVA	[mode -> ErrorIfExists, partitionBy -> []]	null

Konvertera Parquet till Delta

Du kan göra en konvertering på plats från Parquet-formatet till Delta.

Här ska du testa om den befintliga tabellen är i deltaformat eller inte.

parquet_path = "/parquet/parquet-table-{0}".format(session_id)
data = spark.range(0,5)
data.write.parquet(parquet_path)
DeltaTable.isDeltaTable(spark, parquet_path)

var parquetPath = $"/parquet/parquet-table-{sessionId}";
var data = spark.Range(0,5);
data.Write().Parquet(parquetPath);
DeltaTable.IsDeltaTable(parquetPath)

val parquetPath = s"/parquet/parquet-table-$sessionId"
val data = spark.range(0,5)
data.write.parquet(parquetPath)
DeltaTable.isDeltaTable(parquetPath)

Resulterar i:

Falskt

Nu ska du konvertera data till deltaformat och kontrollera att de fungerade.

DeltaTable.convertToDelta(spark, "parquet.`{0}`".format(parquet_path))
DeltaTable.isDeltaTable(spark, parquet_path)

DeltaTable.ConvertToDelta(spark, $"parquet.`{parquetPath}`");
DeltaTable.IsDeltaTable(parquetPath)

DeltaTable.convertToDelta(spark, s"parquet.`$parquetPath`")
DeltaTable.isDeltaTable(parquetPath)

Resulterar i:

Sant

SQL-stöd

Delta stöder tabellverktygskommandon via SQL. Du kan använda SQL för att:

Hämta deltatabellens historik
Dammsuga en DeltaTable
Konvertera en Parquet-fil till Delta

spark.sql("DESCRIBE HISTORY delta.`{0}`".format(delta_table_path)).show()

spark.Sql($"DESCRIBE HISTORY delta.`{deltaTablePath}`").Show();

spark.sql(s"DESCRIBE HISTORY delta.`$deltaTablePath`").show()

Resultat i:

version	timestamp	userId	userName	operation	operationParameters	jobb	notebook-fil	clusterId	readVersion	isolationLevel	isBlindAppend
5	2020-04-25 00:37:09	null	null	DIREKTUPPSPELNINGSUPPDATERING	[outputMode –> Ap...	null	null	null	4	null	true
4	2020-04-25 00:36:27	null	null	SAMMANFOGA	[predikat -> (ol...	null	null	null	3	null	falskt
3	2020-04-25 00:36:08	null	null	DELETE	[predikat -> ["(...	null	null	null	2	null	falskt
2	2020-04-25 00:35:51	null	null	UPDATE	[predikat -> ((i...	null	null	null	1	null	falskt
1	2020-04-25 00:35:05	null	null	SKRIVA	[mode -> Överskrivning...	null	null	null	0	null	falskt
0	2020-04-25 00:34:34	null	null	SKRIVA	[mode -> ErrorIfE...	null	null	null	null	null	true

spark.sql("VACUUM delta.`{0}`".format(delta_table_path)).show()

spark.Sql($"VACUUM delta.`{deltaTablePath}`").Show();

spark.sql(s"VACUUM delta.`$deltaTablePath`").show()

Resultat i:

path
abfss://data@arca...

Nu ska du kontrollera att en tabell inte är en deltaformattabell. Sedan konverterar du tabellen till deltaformat med Spark SQL och bekräftar att den har konverterats korrekt.

parquet_id = random.randint(0,1000)
parquet_path = "/parquet/parquet-table-{0}-{1}".format(session_id, parquet_id)
data = spark.range(0,5)
data.write.parquet(parquet_path)
DeltaTable.isDeltaTable(spark, parquet_path)
spark.sql("CONVERT TO DELTA parquet.`{0}`".format(parquet_path))
DeltaTable.isDeltaTable(spark, parquet_path)

var parquetId =  (new Random()).Next(10000000);
var parquetPath = $"/parquet/parquet-table-{sessionId}-{parquetId}";
var data = spark.Range(0,5);
data.Write().Parquet(parquetPath);
DeltaTable.IsDeltaTable(parquetPath);
spark.Sql($"CONVERT TO DELTA parquet.`{parquetPath}`");
DeltaTable.IsDeltaTable(parquetPath);

val parquetId = scala.util.Random.nextInt(1000)
val parquetPath = s"/parquet/parquet-table-$sessionId-$parquetId"
val data = spark.range(0,5)
data.write.parquet(parquetPath)
DeltaTable.isDeltaTable(parquetPath)
spark.sql(s"CONVERT TO DELTA parquet.`$parquetPath`")
DeltaTable.isDeltaTable(parquetPath)

Resultat i:

Sant

Fullständig dokumentation finns på Delta Lake-dokumentationssidan

Mer information finns i Delta Lake Project.

Dela via

Översikt över Linux Foundation Delta Lake

Konfiguration

Skapa en tabell

Läsa data

Uppdatera tabelldata

Spara som katalogtabeller

Villkorsstyrd uppdatering utan att skriva över

Historik

Läsa äldre versioner av data med time travel

Skriva en dataström till en tabell

Läsa en dataström med ändringar från en tabell

Konvertera Parquet till Delta

SQL-stöd

Nästa steg

Feedback

Feedback

Ytterligare resurser