Apache Hudi 是一种流行的开源 Lakehouse 技术，在大数据社区中迅速发展。如果您在 AWS 上构建了数据湖和数据工程平台，您可能已经听说过或使用过 Apache Hudi，因为 AWS 在其众多数据服务（包括 EMR、Redshift、Athena、Glue 等）中原生集成并支持了 Apache Hudi

在考虑在 Microsoft Azure^[1] 上构建数据湖或 Lakehouse 时，大多数人已经熟悉 Databricks 的 Delta Lake，但现在有些人发现 Apache Hudi 是一种可行的替代方案，可与 Azure Synapse Analytics^[2]、HDInsight^[3]、ADLS Gen2^[4] 、EventHubs^[5] 等 Azure 原生服务一起使用，甚至包括 Azure Databricks。由于可用的文档有限，本博客的目标是分享有关如何在 Azure 上轻松启动和使用 Hudi 的步骤。Apache Hudi 是一种开源 Lakehouse 技术，使你能够将事务、并发、更新插入和高级存储性能优化引入 Azure Data Lake Storage (ADLS) 上的数据湖。Apache Hudi 为您的工作负载提供了显着的性能优势^[6]，并确保您的数据不会被锁定或绑定到任何一家供应商。使用 Apache Hudi，您的所有数据^[7]都保留在 parquet 等开放文件格式中，Hudi 可以与任何流行的查询引擎（如 Apache Spark、Flink、Presto、Trino、Hive 等）一起使用。

Hudi 和 Synapse 设置

在 Azure 上使用 Apache Hudi 就像将一些库加载到 Azure Synapse Analytics 中一样简单。如果您已经拥有 Synapse 工作区、Spark Pool 和 ADLS 存储帐户，则可以跳过以下一些先决条件步骤。

准备条件 – Synapse工作区

首先如果还没有工作区，请创建一个 Synapse 工作区^[8]。无需特殊配置，您需要记住 ADLS 帐户名称和文件系统名称以供以后使用。

准备条件 – Serverless Spark Pool

将 Hudi 与 Apache Spark 一起使用非常容易，因此让我们创建一个 Synapse Serverless Spark池，我们可以使用它来读取和写入数据。首先打开刚刚创建的 Synapse Workspace 并在 Synapse Studio 中启动它。按照此快速入门创建 Spark 池^[9]。您无需为 Hudi 设置特定设置，但请注意您选择的 Spark 运行时版本，并确保您选择了合适的 Hudi 版本^[10]来匹配。在本教程中，我选择了使用 Scala 2.12.10 和 Java 1.8 的 Synapse 中的 Spark 3.1。

准备条件 – 将 Hudi 包添加到 Synapse

虽然 Apache Hudi 库默认预装在 AWS 和 GCP 上的大多数其他数据服务上，但您需要执行一个额外步骤才能在 Azure 上运行 Hudi。有关如何将包安装到 Synapse 工作区的信息，请参阅文档^[11]。使用 Spark 3.1可以配套使用 Hudi 0.10+，可以从 Maven 中心下载了最新的 Hudi 0.11.1 jar：https://search.maven.org/search?q=a:hudi-spark3.1-bundle_2.12 从此处的工作区将 .jar 包上传到 Synapse 工作区：

现在包已上传到工作区，您需要将此包添加到工作区池。导航到Spark池并查找包选项：

单击“从工作区包中选择”

并选择您上传到工作区的 Hudi .jar：

准备条件 – 创建 Synapse Notebook

在 Synapse Studio 中创建一个Notebook^[12]。本教程使用的是 Scala，也可以使用 Python。

Hudi 和 Synapse 快速入门

这个简单的快速入门会让您大吃一惊。有关更深入的材料请参阅下面的其他资源。如果您不想在教程的其余部分复制/粘贴代码，您可以从我的 Github 下载此笔记本^[13]，然后将其导入Synapse 工作区^[14]。

导入样本数据

对于这个简单的快速入门，我使用了来自 AML 开放数据集的经典 NYC Taxi 数据集。

from azureml.opendatasets import NycTlcYellow
from dateutil import parser
from datetime import datetime
 
end_date = parser.parse('2018-06-06')
start_date = parser.parse('2018-05-01')
nyc_tlc = NycTlcYellow(start_date=start_date, end_date=end_date)
filtered_df = nyc_tlc.to_spark_dataframe()
 
display(filtered_df)

设置 Hudi 写入配置

选择一个 Hudi 基本路径并设置基本的写入配置。

• 在 Hudi 文档中阅读有关 Hudi 记录键、预组合键和其他配置的信息：https://hudi.apache.org/docs/writing_data
• 在此处阅读有关 Hudi 写入操作的信息：https://hudi.apache.org/docs/write_operations

basePath = "abfs://kwadlsfilesyshudi@kwadlshudi.dfs.core.windows.net/hudi-test/"
tableName = "hudi_test_data"
 
hudi_options = {
   'hoodie.table.name': tableName,
   'hoodie.datasource.write.recordkey.field': 'tpepPickupDateTime',
   'hoodie.datasource.write.operation': 'upsert',
   'hoodie.datasource.write.precombine.field': 'tpepDropoffDateTime'
}

将样本数据集作为 Hudi 表写入 ADLS G2

只需要使用hudi format即可：spark.write.format("hudi")

filtered_df.write.format("hudi").options(**hudi_options).mode("overwrite").save(basePath)

创建 SQL 表

可以使用 Hudi 关键字创建托管或外部共享表 https://docs.microsoft.com/en-us/azure/synapse-analytics/metadata/table#shared-spark-tables

spark.sql("CREATE TABLE HudiTable USING HUDI LOCATION '{0}'".format(basePath))

现在可以充分利用 Hudi 表上的基本 SQL：

%%sql
select * from HudiTable

Upserts/Merges

Apache Hudi 在数据湖上提供了首个此类高性能索引子系统^[15]。借助记录级索引和 ACID 事务，Hudi 可以轻松快速高效地进行更新插入和合并。假设在出租车行程结束后，乘客决定在行程结束几小时或几天后更改他的小费。对于这个例子，我们先抓取一条记录并检查原始的 tipAmount

origvalue = spark. 
 read. 
 format("hudi"). 
 load(basePath). 
 where("_hoodie_record_key = '1526970901000000'")
 
display(origvalue.select("tipAmount"))

现在将 Hudi 写入操作设置为 upsert，将 tipAmount更改为 $5.23，并将更新后的值写入 append

from pyspark.sql.functions import lit
 
hudi_options = {
   'hoodie.table.name': tableName,
   'hoodie.datasource.write.recordkey.field': 'tpepPickupDateTime',
   'hoodie.datasource.write.operation': 'upsert',
   'hoodie.datasource.write.precombine.field': 'tpepDropoffDateTime'
}
 
updatevalue = origvalue.withColumn("tipAmount",lit(5.23))
 
updatevalue.write.format("hudi"). 
 options(**hudi_options). 
 mode("append"). 
 save(basePath)

原始值变更了

testupdate = spark. 
 read. 
 format("hudi"). 
 load(basePath). 
 where("_hoodie_record_key = '1526970901000000'")
 
display(testupdate.select("tipAmount"))

时间旅行

使用 Apache Hudi可以编写时间旅行查询来重现数据集在某个时间点的样子。可以使用提交瞬间或时间戳指定时间点：https://hudi.apache.org/docs/quick-start-guide#time-travel-query 有几种方法可以找到提交瞬间（查询表详细信息、使用 Hudi CLI 或检查存储）。为简单起见我在 Synapse Studio 中打开了 ADLS 浏览器并导航到保存我的数据的文件夹。该文件夹中有一个 .hoodie 文件夹，其中包含以 [commit instance id].commit 表示的提交列表。下面例子选择了最早的一个。

查询如下

testupdate = spark. 
 read. 
 format("hudi"). 
 option("as.of.instant", "20220624055125356"). 
 load(basePath). 
 where("_hoodie_record_key = '1526970901000000'")
 
display(testupdate.select("tipAmount"))
# you can travel back in time before the upsert and reproduce the original tipAmount

增量查询

Apache Hudi 能够用高效的增量管道替换老式的批处理数据管道。用户可以指定“增量”查询类型，并向 Hudi 询问在给定提交或时间戳之后的所有新记录或更新记录。（对于这个例子，它只是我们更新的那一行，但请随意尝试更多）

# incrementally query data
incremental_read_options = {
 'hoodie.datasource.query.type': 'incremental',
 'hoodie.datasource.read.begin.instanttime': 20220624055125356,
}
 
tripsIncrementalDF = spark.read.format("hudi"). 
 options(**incremental_read_options). 
 load(basePath)
 
display(tripsIncrementalDF)

删除

对于大多数数据湖而言，一项具有挑战性的任务是处理删除，尤其是在处理 GDPR 合规性法规时。Apache Hudi 处理快速高效的删除，同时提供高级并发控制配置。查询要删除的记录

todelete = spark. 
 read. 
 format("hudi"). 
 load(basePath). 
 where("_hoodie_record_key = '1527015638000000'")
 
display(todelete)

接下来将 Hudi 写入操作设置为 delete 并将这些记录作为追加写入：

hudi_delete_options = {
   'hoodie.table.name': tableName,
   'hoodie.datasource.write.recordkey.field': 'tpepPickupDateTime',
   'hoodie.datasource.write.operation': 'delete',
   'hoodie.datasource.write.precombine.field': 'tpepDropoffDateTime'
}
 
todelete.write.format("hudi").options(**hudi_delete_options).mode("append").save(basePath)

可以确认记录已被删除：

todelete = spark. 
 read. 
 format("hudi"). 
 load(basePath). 
 where("_hoodie_record_key = '1527015638000000'")
 
display(todelete)

结论

希望这提供了有关如何开始在 Azure 上使用 Apache Hudi 的指南。虽然本文专注于 Azure Synapse Analytics，但这并不是 Azure 产品组合中唯一可以使用 Apache Hudi 的产品。以下是在使用 Apache Hudi 在 Azure 上开发数据平台时可以考虑的一些架构的想法