Descrição geral do Delta Lake da Fundação Linux

Artigo
08/15/2023

Este artigo foi adaptado para maior clareza por parte do seu homólogo original aqui. Este artigo ajuda-o a explorar rapidamente as principais funcionalidades do Delta Lake. O artigo fornece fragmentos de código que mostram como ler e escrever em tabelas do Delta Lake a partir de consultas interativas, em lote e de transmissão em fluxo. Os fragmentos de código também estão disponíveis num conjunto de blocos de notas PySpark aqui, Scala aqui e C# aqui

Eis o que vamos abordar:

Criar uma tabela
Ler dados
Atualizar dados da tabela
Substituir dados da tabela
Atualização condicional sem substituição
Ler versões mais antigas de dados com o Time Travel
Escrever um fluxo de dados numa tabela
Ler um fluxo de alterações a partir de uma tabela
Suporte do SQL

Configuração

Certifique-se de que modifica o seguinte conforme adequado para o seu ambiente.

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

Resulta em:

"/delta/delta-table-335323"

Criar uma tabela

Para criar uma tabela do Delta Lake, escreva um DataFrame num DataFrame no formato delta. Pode alterar o formato de Parquet, CSV, JSON, etc., para delta.

O código que se segue mostra-lhe como criar uma nova tabela do Delta Lake com o esquema inferido do 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)

Resulta em:

ID
0
1
2
3
4

Ler dados

Pode ler dados na sua tabela do Delta Lake ao especificar o caminho para os ficheiros e o formato delta.

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

Resulta em:

ID
1
3
4
0
2

A ordem dos resultados é diferente de acima, uma vez que não foi especificada explicitamente nenhuma ordem antes de exportar os resultados.

Atualizar dados da tabela

O Delta Lake suporta várias operações para modificar tabelas com APIs de DataFrame padrão. Estas operações são um dos melhoramentos que o formato delta adiciona. O exemplo seguinte executa uma tarefa de lote para substituir os dados na tabela.

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

Resulta em:

ID
7
8
5
9
6

Aqui, pode ver que todos os cinco registos foram atualizados para conter novos valores.

Guardar como tabelas de catálogo

O Delta Lake pode escrever em tabelas de catálogo geridas ou externas.

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

Resulta em:

base de dados	tableName	isTemporary
predefinição	externaldeltatable	false
predefinição	geridodeltatable	false

Com este código, criou uma nova tabela no catálogo a partir de um dataframe existente, referido como uma tabela gerida. Em seguida, definiu uma nova tabela externa no catálogo que utiliza uma localização existente, referida como uma tabela externa. No resultado, pode ver que ambas as tabelas, independentemente da forma como foram criadas, são listadas no catálogo.

Agora, pode ver as propriedades expandidas de ambas as tabelas

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

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

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

Resulta em:

col_name	data_type	comentário
ID	bigint	nulo

Informações Detalhadas da Tabela
Base de Dados	predefinição
Tabela	geridodeltatable
Proprietário	trusted-service-user
Hora de Criação	Sáb 25 00:35:34 UTC 2020
Último Acesso	Qui Jan 01 00:00:00 UTC 1970
Criada Por	Spark 2.4.4.2.6.99.201-11401300
Tipo	GERIDO
Fornecedor	delta
Propriedades da Tabela	[transient_lastDdlTime=1587774934]
Estatísticas	2407 bytes
Localização	abfss://data@<data lake.dfs.core.windows.net/synapse/workspaces/>< workspace name>/warehouse/manageddeltatable
Biblioteca serde	org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
InputFormat	org.apache.hadoop.mapred.SequenceFileInputFormat
OutputFormat	org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat
Propriedades de Armazenamento	[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)

Resultados em:

col_name	data_type	comentário
ID	bigint	nulo

Informações Detalhadas da Tabela
Base de Dados	predefinição
Tabela	externaldeltatable
Proprietário	trusted-service-user
Hora de Criação	Sáb 25 00:35:38 UTC 2020
Último Acesso	Thu Jan 01 00:00:00 UTC 1970
Criada Por	Spark 2.4.4.2.6.99.201-11401300
Tipo	EXTERNO
Fornecedor	DELTA
Propriedades da Tabela	[transient_lastDdlTime=1587774938]
Localização	abfss://data@<data lake.dfs.core.windows.net/delta/delta-table-587152>
Biblioteca serde	org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
InputFormat	org.apache.hadoop.mapred.SequenceFileInputFormat
OutputFormat	org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat
Propriedades de Armazenamento	[serialization.format=1]

Atualização condicional sem substituição

O Delta Lake fornece APIs programáticas para atualização condicional, eliminação e intercalação (este comando é normalmente referido como um upsert) em tabelas.

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

Resultados em:

ID
106
108
5
7
9

Aqui acabou de adicionar 100 a cada ID par.

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

Resultados em:

ID
5
7
9

Repare que todas as linhas pares foram eliminadas.

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

Resultados em:

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

Aqui tem uma combinação dos dados existentes. Os dados existentes foram atribuídos ao valor -1 no caminho de código update(WhenMatched). Os novos dados que foram criados na parte superior do fragmento e que foram adicionados através do caminho de código de inserção (WhenNotMatched), também foram adicionados.

Histórico

O Delta Lake's tem a capacidade de permitir olhar para a história de uma tabela. Ou seja, as alterações efetuadas à Tabela Delta subjacente. A célula abaixo mostra como é simples inspecionar o histórico.

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

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

deltaTable.history.show(false)

Resultados em:

versão	carimbo de data/hora	userId	userName	operation	operationParameters	tarefa	bloco de notas	clusterId	readVersion	isolationLevel	isBlindAppend
4	2020-04-25 00:36:27	nulo	nulo	INTERCALAR	[predicado -> (oldData.`ID` = newData.`ID`)]	nulo	nulo	nulo	3	nulo	false
3	2020-04-25 00:36:08	nulo	nulo	DELETE	[predicado -> ["((`ID` % CAST(2 AS BIGINT)) = CAST(0 AS BIGINT))"]]	nulo	nulo	nulo	2	nulo	false
2	2020-04-25 00:35:51	nulo	nulo	UPDATE	[predicado -> ((ID#744L % cast(2 as bigint)) = cast(0 as bigint))]	nulo	nulo	nulo	1	nulo	false
1	2020-04-25 00:35:05	nulo	nulo	ESCREVER	[modo -> Substituir, partitionBy -> []]	nulo	nulo	nulo	0	nulo	false
0	2020-04-25 00:34:34	nulo	nulo	ESCREVER	[modo -> ErrorIfExists, partitionBy -> []]	nulo	nulo	nulo	nulo	nulo	true

Aqui, pode ver todas as modificações efetuadas nos fragmentos de código acima.

Ler versões mais antigas de dados com o Time Travel

É possível consultar instantâneos anteriores da sua tabela do Delta Lake através de uma funcionalidade denominada Viagem no Tempo. Se quiser aceder aos dados que substituiu, pode consultar um instantâneo da tabela antes de substituir o primeiro conjunto de dados com a opção versionAsOf.

Depois de executar a célula abaixo, deverá ver o primeiro conjunto de dados de antes de a substituir. O Time Travel é uma funcionalidade avançada que tira partido do poder do registo de transações do Delta Lake para aceder a dados que já não estão na tabela. Remover a opção da versão 0 (ou especificar a versão 1) permitir-lhe-ia ver novamente os dados mais recentes. Para obter mais informações, veja Consultar um instantâneo mais antigo de uma tabela.

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

Resulta em:

ID
0
1
4
3
2

Aqui, pode ver que voltou à versão mais antiga dos dados.

Escrever um fluxo de dados numa tabela

Também pode escrever numa tabela do Delta Lake com a Transmissão em Fluxo Estruturada do Spark. O registo de transações do Delta Lake garante exatamente uma vez o processamento, mesmo quando existem outros fluxos ou consultas em lote em execução em simultâneo na tabela. Por predefinição, os fluxos são executados no modo de acréscimo, o que adiciona novos registos à tabela.

Para obter mais informações sobre a integração do Delta Lake com a Transmissão em Fluxo Estruturada, veja Leituras e Escritas de Transmissão em Fluxo de Tabelas.

Nas células abaixo, eis o que estamos a fazer:

Célula 30 Mostrar os dados recém-anexados
Célula 31 Inspecionar histórico
Célula 32 Parar a tarefa de transmissão em fluxo estruturada
Célula 33 Inspecionar histórico <- Verá que os acréscimos pararam

Primeiro, vai configurar uma tarefa simples de Transmissão em Fluxo do Spark para gerar uma sequência e fazer com que a tarefa escreva na sua Tabela Delta.

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)

Ler um fluxo de alterações a partir de uma tabela

Enquanto o fluxo está a escrever na tabela do Delta Lake, também pode ler a partir dessa tabela como uma origem de transmissão em fluxo. Por exemplo, pode iniciar outra consulta de transmissão em fluxo que imprima todas as alterações efetuadas à tabela do Delta Lake.

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

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

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

Resulta em:

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

Resulta em:

versão	carimbo de data/hora	operation	operationParameters	readVersion
5	2020-04-25 00:37:09	ATUALIZAÇÃO DE TRANSMISSÃO EM FLUXO	[outputMode -> Acrescentar, queryId -> d26b4f8a-7e5a-44f2-a5fb-23a7bd02aef7, epochId -> 0]	4
4	2020-04-25 00:36:27	INTERCALAR	[predicado -> (oldData.`id` = newData.`id`)]	3
3	2020-04-25 00:36:08	DELETE	[predicado -> ["((`id` % CAST(2 AS BIGINT)) = CAST(0 AS BIGINT))"]]	2
2	2020-04-25 00:35:51	UPDATE	[predicado -> ((id#744L % cast(2 as bigint)) = cast(0 as bigint))]	1
1	2020-04-25 00:35:05	ESCREVER	[modo -> Substituir, partitionBy -> []]	0
0	2020-04-25 00:34:34	ESCREVER	[modo -> ErrorIfExists, partitionBy -> []]	nulo

Aqui, está a remover algumas das colunas menos interessantes para simplificar a experiência de visualização da vista do histórico.

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

Resulta em:

versão	carimbo de data/hora	operation	operationParameters	readVersion
5	2020-04-25 00:37:09	ATUALIZAÇÃO DE TRANSMISSÃO EM FLUXO	[outputMode -> Acrescentar, queryId -> d26b4f8a-7e5a-44f2-a5fb-23a7bd02aef7, epochId -> 0]	4
4	2020-04-25 00:36:27	INTERCALAR	[predicado -> (oldData.`id` = newData.`id`)]	3
3	2020-04-25 00:36:08	DELETE	[predicado -> ["((`id` % CAST(2 AS BIGINT)) = CAST(0 AS BIGINT))"]]	2
2	2020-04-25 00:35:51	UPDATE	[predicado -> ((id#744L % cast(2 as bigint)) = cast(0 as bigint))]	1
1	2020-04-25 00:35:05	ESCREVER	[modo -> Substituir, partitionBy -> []]	0
0	2020-04-25 00:34:34	ESCREVER	[modo -> ErrorIfExists, partitionBy -> []]	nulo

Converter Parquet em Delta

Pode fazer uma conversão no local do formato Parquet para Delta.

Aqui, irá testar se a tabela existente está ou não no formato delta.

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)

Resulta em:

Falso

Agora, vai converter os dados em formato delta e verificar se funcionaram.

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)

Resulta em:

Verdadeiro

Suporte do SQL

A Delta suporta comandos utilitários de tabela através do SQL. Pode utilizar o SQL para:

Obter o histórico de uma DeltaTable
Aspirar uma DeltaTable
Converter um ficheiro Parquet em 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()

Resultados em:

versão	carimbo de data/hora	userId	userName	operation	operationParameters	tarefa	bloco de notas	clusterId	readVersion	isolationLevel	isBlindAppend
5	2020-04-25 00:37:09	nulo	nulo	ATUALIZAÇÃO DE TRANSMISSÃO EM FLUXO	[outputMode -> Ap...	nulo	nulo	nulo	4	nulo	true
4	2020-04-25 00:36:27	nulo	nulo	INTERCALAR	[predicado -> (ol...	nulo	nulo	nulo	3	nulo	false
3	2020-04-25 00:36:08	nulo	nulo	DELETE	[predicado -> ["(...	nulo	nulo	nulo	2	nulo	false
2	2020-04-25 00:35:51	nulo	nulo	UPDATE	[predicado -> ((i...	nulo	nulo	nulo	1	nulo	false
1	2020-04-25 00:35:05	nulo	nulo	ESCREVER	[modo -> Substituir...	nulo	nulo	nulo	0	nulo	false
0	2020-04-25 00:34:34	nulo	nulo	ESCREVER	[modo -> ErrorIfE...	nulo	nulo	nulo	nulo	nulo	true

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

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

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

Resultados em:

caminho
abfss://data@arca...

Agora, vai verificar se uma tabela não é uma tabela de formato delta. Em seguida, irá converter a tabela em formato delta com o SQL do Spark e confirmar que foi convertida corretamente.

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)

Resultados em:

Verdadeiro

Para obter a documentação completa, consulte a Página de Documentação do Delta Lake

Para obter mais informações, veja Delta Lake Project.

Descrição geral do Delta Lake da Fundação Linux

Configuração

Criar uma tabela

Ler dados

Atualizar dados da tabela

Guardar como tabelas de catálogo

Atualização condicional sem substituição

Histórico

Ler versões mais antigas de dados com o Time Travel

Escrever um fluxo de dados numa tabela

Ler um fluxo de alterações a partir de uma tabela

Converter Parquet em Delta

Suporte do SQL

Passos seguintes

Recursos adicionais