【Azure 架构师学习笔记】- Azure Databricks (16) -- Delta Lake 和 ADLS整合

本文属于【Azure 架构师学习笔记】系列。
本文属于【Azure Databricks】系列。
接上文 【【Azure 架构师学习笔记】- Azure Databricks (15) --Delta Lake 和Data Lake

前言

上文提到了Delta Lake， 但是这是一个概念，如果落实到具体的资源服务上，又会有一定的修改和限制。本文介绍一下Delta Lake如何跟Azure Data Lake Store 整合。

Delta Lake是一个开源框架，可以构建在ADLS之上。ADLS 并不内置事务保障或者Delta Lake提供的性能优化。所以单纯ADLS 很难满足现今的数据需求。

Delta Lake 与ADLS 的优势互补

ADLS 对于“存”操作是很快速的，但是对于“搜索、list”文件则很慢。Delta Lake通过减少昂贵的文件列举操作而加快数据检索。
所有的云存储系统使用扁平的命名空间。不像文件系统的层级结构，所以在list文件时很慢，就类似没有索引的数据库中的数据表，需要遍历所有数据。 ADLS 没有真正的目录，而是在对象名字前面添加前缀来实现。所以list 文件意味着系统要扫描全部文件，并且用前缀筛选。

另外， ADLS 上如果受到API 过度调用， 则会出现瓶颈，特别是对大数据集。如果一个大表的数据被拆分到ADLS 上的很多文件，那么ADLS 就要遍历所有文件，然后筛选，从而导致额外的延时。

对此，Delta Lake把文件路径存储在一个事务日志文件中，就如数据库中的索引，list文件时先从事务日志文件中查找，然后再去具体的路径中检索数据。

使用演示

下面使用Spark来演示一下如何操作Delta Lake，这里借用前面的ADB 环境，ADB 实际上非必须，只是借用上面的Spark环境。

环境配置

使用下面python命令配置Spark和Delta， 替换代码中的中文字部分即可。

%python
import pyspark
from delta import *

extra_packages = [
    "org.apache.hadoop:hadoop-azure:3.3.4",
    # version must match hadoop-client-runtime
]

builder = (
    pyspark.sql.SparkSession.builder.appName("DeltaOnADLS")
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config("spark.hadoop.fs.azurebfs.impl", "org.apache.hadoop.fs.azurebfs.AzureBlobFileSystem")
    .config("fs.azure.account.key.ADLS 名字.dfs.core.windows.net", "你的ADLS 的access key")
)

spark = configure_spark_with_delta_pip(
    builder, extra_packages=extra_packages
).getOrCreate()

如果过程中报这样的错： “org.apache.hadoop.fs.FileAlreadyExistsException: Operation failed: “This endpoint does not support BlobStorageEvents or SoftDelete. Please disable these account features if you would like to use this endpoint.”
”
需要禁用ADLS 中这个设置，因为它影响了Delta的ACID 特性。

写入数据

# Create sample data
data = spark.range(0, 5)

# Write data to a Delta table in ADLS
data.write.format("delta").save("abfss://[e-mail]/delta-table")

写入后从ADLS 上可以看到新建了一个文件夹：

文件夹内部是具有transaction log (这里是_delta_log)的文件集合。

读取数据

执行下面命令，注意修改路径为上面写入数据的路径和文件夹

# Read Delta table from ADLS
df = spark.read.format("delta").load("abfss://[e-mail]/delta-table")
df.show()

结果如下：

更新和删除数据

from delta.tables import DeltaTable

# Load Delta table
deltaTable = DeltaTable.forPath(spark, "abfss://[e-mail]/delta-table")

# Update rows where id is less than 4
deltaTable.update("id < 4", {"id": "id + 20"})

# Delete rows where id is less than 5
deltaTable.delete("id < 5")

# Read Delta table from ADLS
df = spark.read.format("delta").load("abfss://[e-mail]/delta-table")
df.show()

结果如下：

Merge数据

# Merge new data into Delta table
new_data = spark.range(0, 30)

deltaTable.alias("old").merge(
    new_data.alias("new"),
    "old.id = new.id"
).whenMatchedUpdate(set={"id": "new.id"}).whenNotMatchedInsert(values={"id": "new.id"}).execute()

# Read Delta table from ADLS
df = spark.read.format("delta").load("abfss://[e-mail]/delta-table")
df.show()

小结

本文演示了如何在ADLS 上搭建delta并进行了简单的数据操作。