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条
文章来源：https://www.toymoban.com/news/detail-807577.html
执行如下语句：停止Steam文章来源地址https://www.toymoban.com/news/detail-807577.html
deltastream.stop()到了这里，关于Data Bricks Delta Lake 入门的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
