Data Bricks Delta Lake 入门

Delta Lake 是一个开源存储层，它将关系数据库语义添加到基于 Spark 的数据湖处理中。适用于 PySpark、Scala 和 .NET 代码的 Azure Synapse Analytics Spark , Azure DataBricks 都支持 Delta Lake。在大数据这个领域，对象存储的最影响效率的问题就是针对对象存储数据的更新，传统的对象存储如AWS 的S3 ， Azure的 Blob等如果要更新要给对象数据的时候，必须要先将对象查找到并删除，然后再追加。这通常会导致性能效率低下。Delta Lake很高的的解决了对象数据更新的问题，并同时支持实时数据流的更新，主要功能如下：

支持查询和数据修改的关系表。使用 Delta Lake，可以将数据存储在支持 CRUD（创建、读取、更新和删除）操作的表中。换句话说，可以采用与在关系数据库系统中相同的方式选择、插入、更新和删除数据行。
支持 ACID 事务。关系数据库旨在支持事务数据修改，这些修改提供原子性（事务作为单个工作单元完成）、一致性（事务使数据库保持一致状态）、隔离（进行中的事务不能相互干扰）和持久性（事务完成时，它所做的更改将保留）。 Delta Lake 通过实现事务日志并强制实施并发操作的可序列化隔离，为 Spark 提供相同的事务支持。
数据版本控制和按时间顺序查看。由于所有事务都记录在事务日志中，因此可以跟踪每个表行的多个版本，甚至使用按时间顺序查看功能在查询中检索某行的先前版本。
支持批处理和流式处理数据。虽然大多数关系数据库包括存储静态数据的表，但 Spark 包含通过 Spark 结构化流式处理 API 流式处理数据的本机支持。 Delta Lake 表可用作流式处理数据的接收器（目标）和源。
标准格式和互操作性。 Delta Lake 表的基础数据以 Parquet 格式存储，该格式通常用于数据湖引入管道。

以下开始Delta的入门操作：

使用免费的Azure Data Bricks 的工作区，参加如下链接：

利用 Azure Data Bricks的免费资源学习云上大数据-CSDN博客

一、创建 Delta Lake 表

1、从数据帧创建 Delta Lake 表

创建Ddelta Lake表，以增量格式保存数据帧，指定应存储表的数据文件和相关元数据信息的路径

下载试验用数据：

 %sh
 rm -r /dbfs/delta_lab
 mkdir /dbfs/delta_lab
 wget -O /dbfs/delta_lab/products.csv https://raw.githubusercontent.com/MicrosoftLearning/mslearn-databricks/main/data/products.csv

例如：使用现有文件中的数据加载数据帧，然后将该数据帧以增量格式保存到新文件夹位置：

# Load a file into a dataframe
df = spark.read.load('/delta_lab/products.csv', format='csv', header=True)

# Save the dataframe as a delta table
delta_table_path = "/delta/mydata"
df.write.format("delta").save(delta_table_path)

2、验证保存的Delta Lake数据文件

保存 delta 表后，指定的路径位置包括数据的 parquet 文件

%sh 
ls /dbfs/delta/mydata

执行后结果如下：

可以使用覆盖模式将现有 Delta Lake 表替换为数据帧的内容，如下所示：

new_df.write.format("delta").mode("overwrite").save(delta_table_path)

还可以使用追加模式将数据帧中的行添加到现有表：

new_rows_df.write.format("delta").mode("append").save(delta_table_path)

二、根据条件进行表的更新

1、首先查看原始表的数据：

df1 = df.select("ProductName", "ListPrice").where((df["ProductName"]=="Road-750 Black, 58"))
display(df1)

执行结果：

2、执行数据更新语句

虽然可以在数据帧中进行数据修改，然后通过覆盖数据来替换 Delta Lake 表，但数据库中的一种更常见的模式是插入、更新或删除现有表中的行作为离散事务操作。若要对 Delta Lake 表进行此类修改，可以使用 Delta Lake API 中的 DeltaTable 对象，该对象支持更新、删除和合并操作。例如，可以使用以下代码更新 category 列值为“Accessories”的所有行的 price 列：

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

# Create a deltaTable object
deltaTable = DeltaTable.forPath(spark, delta_table_path)

# Update the table (reduce price of accessories by 10%)
deltaTable.update(
    condition = "ProductName=='Road-750 Black, 58'",
    set = { "ListPrice": "ListPrice * 0.9" })

执行结果：

3、查看验证数据是否更新：

df2 = spark.read.load('/delta/mydata')
df3= df2.select("ProductName", "ListPrice").where((df2["ProductName"]=="Road-750 Black, 58"))
display(df3)

执行结果：

三、查询表以前的版本

Delta Lake 表支持通过事务日志进行版本控制。事务日志记录对表进行的修改，指出每个事务的时间戳和版本号。可以使用此记录的版本数据查看表以前的版本 - 称为按时间顺序查看的功能。

可以通过将数据从 delta 表位置读取到数据帧中，将所需版本指定为 versionAsOf 选项，从 Delta Lake 表的特定版本检索数据：

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

执行结果：

可以使用 timestampAsOf 选项指定时间戳：

df4 = spark.read.format("delta").option("timestampAsOf", '2024-01-16 09:36:23').load(delta_table_path)

查看历史变化

deltaTable.history(10).show(10, False, True)

执行结果：

四、创建和查询目录表

下面将Delta Lake 表定义为 Spark 群集的 Hive 元存储中的目录表，并使用 SQL 来进行处理。

Spark 目录中的表（包括 Delta Lake 表）可以是托管或外部表（非托管表）

托管表是在没有指定位置的情况下定义的，数据文件存储在元存储使用的存储中。删除表不仅会从目录中删除其元数据，还删除存储其数据文件的文件夹。
外部表指的是为自定义文件位置定义外部表，其中存储了表的数据。表的元数据定义在 Spark 目录中。删除表会从目录中删除元数据，但不会影响数据文件。

1、使用数据帧来创建目录表

可以使用 saveAsTable 操作写入数据帧来创建托管表和非托管表，如以下示例所示：

# 托管表 Save a dataframe as a managed table 
df.write.format("delta").saveAsTable("MyManagedTable")

# 非托管表 specify a path option to save as an external table
df.write.format("delta").option("path", "/mydata").saveAsTable("MyExternalTable")

2、使用 SQL 创建目录表

可以使用含 USING DELTA 子句的 CREATE TABLE SQL 语句和用于外部表的可选 LOCATION 参数来创建目录表。可以使用 SparkSQL API 运行语句，如以下示例所示：

使用Spark SQL 创建一个外部表

spark.sql("CREATE DATABASE AdventureWorks")
spark.sql("CREATE TABLE AdventureWorks.ProductsExternal USING DELTA LOCATION '{0}'".format(delta_table_path))
spark.sql("DESCRIBE EXTENDED AdventureWorks.ProductsExternal").show(truncate=False)

执行结果：

使用SQL直接查询

%sql
USE AdventureWorks;
SELECT * FROM ProductsExternal;

执行结果：

创建一个完全托管的表

df.write.format("delta").saveAsTable("AdventureWorks.ProductsManaged")
spark.sql("DESCRIBE EXTENDED AdventureWorks.ProductsManaged").show(truncate=False)

查询该表：

%sql
USE AdventureWorks;
SELECT * FROM ProductsManaged;

执行结果：

3、比较外部表和托管表的差别

执行如下语句：

%sql
USE AdventureWorks;
SHOW TABLES;

结果如下：

执行下面语句来看两个表所在的不同位置：

 %sh
 echo "External table:"
 ls /dbfs/delta/mydata
 echo
 echo "Managed table:"
 ls /dbfs/user/hive/warehouse/adventureworks.db/productsmanaged

执行结果：

执行下面命令将表删除，看有何不同

%sql
USE AdventureWorks;
DROP TABLE IF EXISTS ProductsExternal;
DROP TABLE IF EXISTS ProductsManaged;
SHOW TABLES;

 %sh
 echo "External table:"
 ls /dbfs/delta/mydata
 echo
 echo "Managed table:"
 ls /dbfs/user/hive/warehouse/adventureworks.db/productsmanaged

执行结果如下：Managed Table的数据文件已经不在，外部表的数据文件还在

五、使用 Delta Lake 对数据进行流式处理

假设我们是一个IoT设备的流式数据，数据结构如下：

1、下载JSON格式的流式数据文件

 %sh
 rm -r /dbfs/device_stream
 mkdir /dbfs/device_stream
 wget -O /dbfs/device_stream/devices1.json https://raw.githubusercontent.com/MicrosoftLearning/mslearn-databricks/main/data/devices1.json

执行结果如下：

下面基于JSON文件所在的文件夹创建iotstream，如下命令：

from pyspark.sql.types import *
from pyspark.sql.functions import *
   
# Create a stream that reads data from the folder, using a JSON schema
inputPath = '/device_stream/'
jsonSchema = StructType([
StructField("device", StringType(), False),
StructField("status", StringType(), False)
])
iotstream = spark.readStream.schema(jsonSchema).option("maxFilesPerTrigger", 1).json(inputPath)
print("Source stream created...")

执行结果：

将数据流写入delta表

# Write the stream to a delta table
delta_stream_table_path = '/delta/iotdevicedata'
checkpointpath = '/delta/checkpoint'
deltastream = iotstream.writeStream.format("delta").option("checkpointLocation", checkpointpath).start(delta_stream_table_path)
print("Streaming to delta sink...")

如下图：数据作业已经启动，开始实时写入

读取实时表中的数据：

# Read the data in delta format into a dataframe
df = spark.read.format("delta").load(delta_stream_table_path)
display(df)

执行结果： 9条记录

基于这个stream创建一个目录表

# create a catalog table based on the streaming sink
spark.sql("CREATE TABLE IotDeviceData USING DELTA LOCATION '{0}'".format(delta_stream_table_path))

查询这张表：

%sql
SELECT * FROM IotDeviceData;

执行结果如下：9条记录

执行以下语句刷新Iot数据到Stream

 %sh
 wget -O /dbfs/device_stream/devices2.json https://raw.githubusercontent.com/MicrosoftLearning/mslearn-databricks/main/data/devices2.json

执行结果：

查看数据是否已经更新到表中，执行如下语句：

%sql
SELECT * FROM IotDeviceData;

执行结果如下：从9条增加到16条

执行如下语句：停止Steam

deltastream.stop()