本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。
# Hudi
[Apache Hudi](https://hudi.apache.org/) 是一种开源数据管理框架，用于通过提供记录级插入、更新、更新插入和删除功能来简化增量数据处理和数据管道开发工作。*更新插入*指的是将记录插入到现有数据集中（如果它们不存在）或对数据集进行更新（如果它们存在）的功能。通过高效地管理数据在 Amazon S3 中的布局方式，Hudi 允许近乎实时地摄取和更新数据。Hudi 仔细维护对数据集执行的操作的元数据，以帮助确保操作是原子级且是一致的。
Hudi 集成了 [Apache Spark](https://aws.amazon.com/emr/features/spark/)、[Apache Hive](https://hive.apache.org/) 和 [Presto](https://prestodb.github.io)。在 Amazon EMR 发行版 6.1.0 及更高版本中，Hudi 还与 [Trino（PrestoSQL）](https://trino.io/)集成。
在 Amazon EMR 5.28.0 版本及更高版本中，EMR 默认情况下会在安装 Spark、Hive、Presto 或 Flink 时安装 Hudi 组件。你可以使用 Spark 或 Hudi DeltaStreamer 实用程序来创建或更新 Hudi 数据集。您可以使用 Hive、Spark、Presto 或 Flink 以交互方式查询 Hudi 数据集，或使用*增量拉取*功能构建数据处理管道。增量拉取是指仅拉取两个操作之间更改的数据的功能。
这些功能使得 Hudi 适用于以下使用案例：
+ 处理来自传感器和其它需要特定数据插入和更新事件的物联网 (IoT) 设备的流数据。
+ 在用户可能会选择被忘记或修改其对数据使用方式的同意的应用程序中，遵守数据隐私法规。
+ 实施[更改数据捕获 (CDC) 系统](https://en.wikipedia.org/wiki/Change_data_capture)，该系统允许您随着时间的推移将更改应用于数据集。
下表列出了 Amazon EMR 7.x 系列的最新发行版附带的 Hudi 版本，以及 Amazon EMR 随 Hudi 一起安装的组件。
有关此版本中与 Hudi 一起安装的组件的版本，请参阅 [7.12.0 版本](emr-7120-release.md)的组件版本。
**emr-7.12.0 的 Hudi 版本信息**
| Amazon EMR 发行版标签 | Hudi 版本 | 随 Hudi 安装的组件 |
| --- | --- | --- |
| emr-7.12.0 | Hudi 1.0.2-amzn-1 | Not available. |
下表列出了 Amazon EMR 6.x 系列的最新发行版附带的 Hudi 版本，以及 Amazon EMR 随 Hudi 一起安装的组件。
有关此发行版中随 Hudi 安装的组件版本，请参阅[发行版 6.15.0 组件版本](emr-6150-release.md)。
**emr-6.15.0 的 Hudi 版本信息**
| Amazon EMR 发行版标签 | Hudi 版本 | 随 Hudi 安装的组件 |
| --- | --- | --- |
| emr-6.15.0 | Hudi 0.14.0-amzn-0 | Not available. |
**注意**
Amazon EMR 发行版 6.8.0 随附 [Apache Hudi](https://hudi.apache.org/) 0.11.1；但是，Amazon EMR 6.8.0 集群也与 Hudi 0.12.0 中的开源 `hudi-spark3.3-bundle_2.12` 兼容。
下表列出了 Amazon EMR 5.x 系列的最新发行版附带的 Hudi 版本，以及 Amazon EMR 随 Hudi 一起安装的组件。
有关此发行版中随 Hudi 安装的组件版本，请参阅[发行版 5.36.2 组件版本](emr-5362-release.md)。
**emr-5.36.2 的 Hudi 版本信息**
| Amazon EMR 发行版标签 | Hudi 版本 | 随 Hudi 安装的组件 |
| --- | --- | --- |
| emr-5.36.2 | Hudi 0.10.1-amzn-1 | Not available. |
**Topics**
+ [Hudi 的工作原理](emr-hudi-how-it-works.md)
+ [在 Amazon EMR 上使用 Hudi 的注意事项和限制](emr-hudi-considerations.md)
+ [创建安装了 Hudi 的集群](emr-hudi-installation-and-configuration.md)
+ [使用 Hudi 数据集](emr-hudi-work-with-dataset.md)
+ [使用 Hudi CLI](emr-hudi-cli.md)
+ [Hudi 发行版历史记录](Hudi-release-history.md)
# Hudi 的工作原理
将 Hudi 与 Amazon EMR 配合使用时，您可以使用 Spark 数据源 API 或 Hudi 实用工具向数据集写入数据。 DeltaStreamer Hudi 将数据集整理到 `basepath` 下一个分区的目录结构中，类似于传统的 Hive 表。如何将数据布局为这些目录中的文件的具体细节取决于您选择的数据集类型。您可以选择“写入时复制 (CoW)”或“读取时合并 (MOM)”。
无论数据集类型如何，数据集中的每个分区都由其相对于 `basepath` 的 `partitionpath` 唯一标识。在每个分区中，记录分布到多个数据文件中。有关更多信息，请参阅 Apache Hudi 文档中的[文件管理](https://hudi.apache.org/docs/concepts.html#file-management)。
Hudi 中的每个操作都有一个相应的提交，由一个单调递增的时间戳标识，称为 *Instant*。Hudi 将对数据集执行的一系列操作保留为时间轴。Hudi 依靠此时间轴提供读取器和写入器之间的快照隔离，并支持回滚到前一个时间点。有关 Hudi 记录的操作和操作状态的更多信息，请参阅 Apache Hudi 文档中的 [Timeline](https://hudi.apache.org/docs/concepts.html#timeline)。
## 了解数据集存储类型：写入时复制与读取时合并
创建 Hudi 数据集时，可以指定数据集在写入时复制或读取时合并。
+ **写入时复制（CoW）**：数据以列状格式存储（Parquet），并且每次更新都会在写入过程中创建一个新版本的文件。CoW 是默认存储类型。
+ **读取时合并（MOR）**：数据使用列式（Parquet）和基于行（Avro）的格式的组合进行存储。更新记录到基于行的*增量*文件中，并根据需要进行压缩以创建新版本的列式文件。
对于 CoW 数据集，每次更新记录时，包含该记录的文件都会使用更新后的值进行重写。对于 MoR 数据集，每次进行更新时，Hudi 仅写入已更改记录对应的行。MoR 更适合写入或更改繁重而读取量较少的工作负载。CoW 更适合更改频率较低但读取量繁重的工作负载。
Hudi 为访问数据提供三个逻辑视图：
+ **读取优化视图**：提供来自 CoW 表的最新提交数据集和来自 MOR 表的最新压缩数据集。
+ **增量视图**：提供 CoW 数据集中两个操作之间的更改流，以馈送给下游作业和提取、转换、加载（ETL）工作流。
+ **实时视图**：通过内联合并列式和基于行的文件，从 MOR 表中提供最新提交的数据。
当您查询读取优化的视图时，查询将返回所有压缩数据，但不包括最新的增量提交。查询此数据可提供良好的读取性能，但忽略最新的数据。当您查询实时视图时，Hudi 会在读取时将压缩的数据与增量提交合并。最新的数据可用于查询，但合并的计算开销使查询性能降低。通过查询压缩数据或实时数据的功能，您可以在查询时在性能和灵活性之间进行选择。
有关在存储类型之间权衡的更多信息，请参阅 Apache Hudi 文档中的[存储类型和视图](https://hudi.apache.org/docs/concepts.html#storage-types--views)。
在 MoR 的 Hive 元数据仓中创建两个表：一个具有您指定的名称的表（即读取优化视图）和一个附加了 `_rt` 的同名表（即实时视图）。您可以查询这两个表。
## 将 Hudi 数据集注册到您的元数据仓
当您向 Hive 元数据仓注册 Hudi 表时，您可以像对待任何其它表一样，使用 Hive、Spark SQL 或 Presto 查询 Hudi 表。此外，你可以将 Hive 和 Spark 配置为使用 AWS Glue 数据目录作为元数据库，从而将 Hudi 与 AWS Glue 集成。对于 MoR 表，Hudi 将数据集注册为元数据仓中的两个表：一个具有您指定的名称的表（即读取优化视图）和一个附加了 `_rt` 的同名表（即实时视图）。
当您使用 Spark 创建 Hudi 数据集时，您可以通过将 `HIVE_SYNC_ENABLED_OPT_KEY` 选项设置为 `"true"` 并提供其它必需的属性来向 Hive 元数据仓注册 Hudi 表。有关更多信息，请参阅[使用 Hudi 数据集](emr-hudi-work-with-dataset.md)。此外，您可以使用 hive\$1sync\$1tool 命令行实用程序将 Hudi 数据集单独注册为元数据仓中的表。
# 在 Amazon EMR 上使用 Hudi 的注意事项和限制
+ **记录键字段不能为 Null 或空**：您指定为记录键字段的字段不能具有 `null` 或空值。
+ **默认情况下，架构会在 upsert 和 insert 时更新** — Hudi 提供了一个接口`HoodieRecordPayload`，用于确定如何合并输入 DataFrame 和现有 Hudi 数据集以生成新的、更新的数据集。Hudi 提供了该类的默认实现`OverwriteWithLatestAvroPayload`，它会覆盖现有记录并更新输入中指定的架构。 DataFrame要自定义此逻辑以实现合并和部分更新，您可以使用 `DataSourceWriteOptions.PAYLOAD_CLASS_OPT_KEY` 参数提供 `HoodieRecordPayload` 接口的实现。
+ **删除需要架构**：删除时，必须指定记录键、分区键和预组合键字段。其它列可以成为 `null` 或空，但需要完整的架构。
+ **MoR 表限制**：MoR 表不支持保存点。您可以使用来自 Spark SQL、Presto 或 Hive 的读取优化视图或实时视图（`tableName_rt`）。使用读取优化视图仅公开基本文件数据，不会公开基本数据和日志数据的合并视图。
+ **Hive**
+ 要在 Hive 元数据仓中注册表，Hudi 需要 Hive Thrift 服务器在默认端口 `10000` 上运行。如果使用自定义端口覆盖此端口，请传递 `HIVE_URL_OPT_KEY` 选项，如以下示例所示。
```
.option(DataSourceWriteOptions.HIVE_URL_OPT_KEY, "jdbc:hive2://localhost:override-port-number
```
+ Spark 中的 `timestamp` 数据类型在 Hive 中注册为 `long` 数据类型，而不注册为 Hive 的 `timestamp` 类型。
+ **：Presto**
+ 在版本低于 0.6.0 的 Hudi 中，Presto 不支持读取 MoR 实时表。
+ Presto 仅支持快照查询。
+ 要使 Presto 正确解释 Hudi 数据集列，请将 `hive.parquet_use_column_names` 值设置为 `true`。
+ 要设置会话的值，请在 Presto shell 中运行以下命令：
```
set session hive.parquet_use_column_names=true
```
+ 要在集群级别设置值，请使用 `presto-connector-hive` 配置分类将 `hive.parquet.use_column_names` 设置为 `true`，如以下示例所示。有关更多信息，请参阅 [配置应用程序](emr-configure-apps.md)。
```
[
{
"Classification": "presto-connector-hive",
"Properties": {
"hive.parquet.use-column-names": "true"
}
}
]
```
+ **HBase 索引**
+ 用于*构建* Hudi 的 HBase 版本可能与 EMR 发行指南中列出的版本不同。要为 Spark 会话提取正确的依赖项，请运行以下命令。
```
spark-shell \
--jars /usr/lib/spark/external/lib/spark-avro.jar,/usr/lib/hudi/cli/lib/*.jar \
--conf "spark.serializer=org.apache.spark.serializer.KryoSerializer" \
--conf "spark.sql.hive.convertMetastoreParquet=false"
```
+ **最佳性能设置**：对于 EMR 7.3\$1/Hudi 0.15\$1，建议客户设置此配置以减少 Kryo 序列化开销：
```
--conf 'spark.kryo.registrator=org.apache.spark.HoodieKryoRegistrar'
```
**注意**
如果您在 EMR Serverless 上使用细粒度访问控制 (FGAC)，则不需要此配置，因为用户必须使用而不是。 JavaSerializer KryoSerializer
# 创建安装了 Hudi 的集群
在 Amazon EMR 版本 5.28.0 及更高版本中， Amazon EMR 默认情况下会在安装 Spark、Hive 或 Presto 时安装 Hudi 组件。要在 Amazon EMR 上使用 Hudi，请在安装了以下一个或多个应用程序后创建集群：
+ Hadoop
+ Hive
+ Spark
+ ：Presto
+ Flink
您可以使用 AWS 管理控制台 AWS CLI、或 Amazon EMR API 创建集群。
## 要使用 Hudi 创建集群，请使用 AWS 管理控制台
1. 导航到 Amazon EMR 新控制台，然后从侧面导航栏中选择**切换到旧控制台**。有关切换到旧控制台后预期情况的更多信息，请参阅 [Using the old console](https://docs.aws.amazon.com/emr/latest/ManagementGuide/whats-new-in-console.html#console-opt-in)。
1. 依次选择 **Create cluster (创建集群)**、**Go to advanced options (转到高级选项)**。
1. 在“软件配置”下，对于**发行版**选择 **emr-5.28.0** 或更高版本，然后选择 **Hadoop**、**Hive**、**Spark**、**Presto** 和 **Tez** 以及集群需要的其他应用程序。
1. 根据应用程序的需要配置其它选项，然后选择**下一步**。
1. 根据需要配置 **Hardware (硬件)** 和 **General cluster settings (常规集群设置)** 的选项。
1. 对于 **Security Options (安全选项)**，我们建议您选择一个 **EC2 key pair (EC2 密钥对)**，您可以使用它通过 SSH 连接到主节点命令行。这允许您运行本指南中描述的 Spark shell 命令、Hive CLI 命令和 Hudi CLI 命令。
1. 根据需要选择其它安全选项，然后选择 **Create cluster (创建集群)**。
# 使用 Hudi 数据集
Hudi 支持通过 Spark 在 Hudi 数据集中插入、更新和删除数据。有关更多信息，请参阅 Apache Hudi 文档中的[写入 Hudi 表格](https://hudi.apache.org/docs/writing_data.html)。
以下示例演示如何启动交互式 Spark Shell、使用 Spark 提交，或如何使用 Amazon EMR Notebooks 在 Amazon EMR 上使用 Hudi。您也可以使用 Hudi DeltaStreamer 实用程序或其他工具写入数据集。在本节中，示例演示使用 Spark shell 处理数据集，同时使用 SSH 作为默认 `hadoop` 用户连接到主节点。
## 使用 Amazon EMR 6.7 及更高版本启动 Spark Shell
运行 `spark-shell`、`spark-submit` 或 `spark-sql` 使用 Amazon EMR 6.7.0 或更高版本时，传递以下命令。
**注意**
Amazon EMR 6.7.0 使用 [Apache Hudi 0.11.0-amzn-0](https://hudi.apache.org/)，相比于之前的 Hudi 版本有明显改进。有关更多信息，请参阅 [Apache Hudi 0.11.0 Migration Guide](https://hudi.apache.org/releases/release-0.11.0/#migration-guide)（《Apache Hudi 0.11.0 迁移指南》）。此选项卡上的示例反映了这些更改。
**在主节点上打开 Spark Shell**
1. 使用 SSH 连接到主节点。有关更多信息，请参阅《Amazon EMR 管理指南》**中的[使用 SSH 连接到主节点](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-connect-master-node-ssh.html)。
1. 输入以下命令以启动 Spark shell。要使用 PySpark 外壳，请*spark-shell*替换为*pyspark*。
```
spark-shell --jars /usr/lib/hudi/hudi-spark-bundle.jar \
--conf "spark.serializer=org.apache.spark.serializer.KryoSerializer" \
--conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog" \
--conf "spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension"
```
## 使用 Amazon EMR 6.6 及更早版本启动 Spark Shell
运行 `spark-shell`、`spark-submit` 或 `spark-sql` 使用 Amazon EMR 6.6.x 或更早版本时，传递以下命令。
**注意**
Amazon EMR 6.2 和 5.31 及更高版本（Hudi 0.6.x 及更高版本）可以在配置中省略 `spark-avro.jar`。
Amazon EMR 6.5 和 5.35 及更高版本（Hudi 0.9.x 及更高版本）可以从配置中省略 `spark.sql.hive.convertMetastoreParquet=false`。
Amazon EMR 6.6 和 5.36 及更高版本（Hudi 0.10.x 及更高版本）必须包含 [Version: 0.10.0 Spark Guide](https://hudi.apache.org/docs/0.10.0/quick-start-guide/)（《版本：0.10.0 Spark 指南》）中所述的 `HoodieSparkSessionExtension` 配置：
```
--conf "spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension" \
```
**在主节点上打开 Spark Shell**
1. 使用 SSH 连接到主节点。有关更多信息，请参阅《Amazon EMR 管理指南》**中的[使用 SSH 连接到主节点](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-connect-master-node-ssh.html)。
1. 输入以下命令以启动 Spark shell。要使用 PySpark 外壳，请*spark-shell*替换为*pyspark*。
```
spark-shell \
--conf "spark.serializer=org.apache.spark.serializer.KryoSerializer" \
--conf "spark.sql.hive.convertMetastoreParquet=false" \
--jars /usr/lib/hudi/hudi-spark-bundle.jar,/usr/lib/spark/external/lib/spark-avro.jar
```
## 使用 Amazon EMR 6.7 及更高版本将 Hudi 与 Amazon EMR Notebooks 结合使用
要将 Hudi 与 Amazon EMR Notebooks 结合使用，您必须首先将 Hudi jar 文件从本地文件系统复制到 Notebook 集群的主节点 上的 HDFS。然后，您可以使用 Notebook 编辑器来配置 EMR Notebook 以使用 Hudi。
**将 Hudi 与 Amazon EMR Notebooks 搭配使用**
1. 为 Amazon EMR Notebooks 创建并启动集群。有关更多信息，请参阅《Amazon EMR 管理指南》**中的[为 Notebook 创建 Amazon EMR 集群](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-managed-notebooks-cluster.html)。
1. 使用 SSH 连接到集群的主节点，然后将 jar 文件从本地文件系统复制到 HDFS，如以下示例所示。在此示例中，我们在 HDFS 中创建了一个目录，以便清晰地管理文件。如果需要，您可以在 HDFS 中选择自己的目的地。
```
hdfs dfs -mkdir -p /apps/hudi/lib
```
```
hdfs dfs -copyFromLocal /usr/lib/hudi/hudi-spark-bundle.jar /apps/hudi/lib/hudi-spark-bundle.jar
```
1. 打开 Notebook 编辑器，输入以下示例中的代码，然后运行它。
```
%%configure
{ "conf": {
"spark.jars":"hdfs:///apps/hudi/lib/hudi-spark-bundle.jar",
"spark.serializer":"org.apache.spark.serializer.KryoSerializer",
"spark.sql.catalog.spark_catalog": "org.apache.spark.sql.hudi.catalog.HoodieCatalog",
"spark.sql.extensions":"org.apache.spark.sql.hudi.HoodieSparkSessionExtension"
}}
```
## 使用 Amazon EMR 6.6 及更早版本将 Hudi 与 Amazon EMR Notebooks 结合使用
要将 Hudi 与 Amazon EMR Notebooks 结合使用，您必须首先将 Hudi jar 文件从本地文件系统复制到 Notebook 集群的主节点 上的 HDFS。然后，您可以使用 Notebook 编辑器来配置 EMR Notebook 以使用 Hudi。
**将 Hudi 与 Amazon EMR Notebooks 搭配使用**
1. 为 Amazon EMR Notebooks 创建并启动集群。有关更多信息，请参阅《Amazon EMR 管理指南》**中的[为 Notebook 创建 Amazon EMR 集群](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-managed-notebooks-cluster.html)。
1. 使用 SSH 连接到集群的主节点，然后将 jar 文件从本地文件系统复制到 HDFS，如以下示例所示。在此示例中，我们在 HDFS 中创建了一个目录，以便清晰地管理文件。如果需要，您可以在 HDFS 中选择自己的目的地。
```
hdfs dfs -mkdir -p /apps/hudi/lib
```
```
hdfs dfs -copyFromLocal /usr/lib/hudi/hudi-spark-bundle.jar /apps/hudi/lib/hudi-spark-bundle.jar
```
```
hdfs dfs -copyFromLocal /usr/lib/spark/external/lib/spark-avro.jar /apps/hudi/lib/spark-avro.jar
```
1. 打开 Notebook 编辑器，输入以下示例中的代码，然后运行它。
```
{ "conf": {
"spark.jars":"hdfs:///apps/hudi/lib/hudi-spark-bundle.jar,hdfs:///apps/hudi/lib/spark-avro.jar",
"spark.serializer":"org.apache.spark.serializer.KryoSerializer",
"spark.sql.hive.convertMetastoreParquet":"false"
}}
```
## 初始化 Hudi 的 Spark 会话
使用 Scala 时，您必须在 Spark 会话中导入以下类。这需要在每个 Spark 会话中完成一次。
```
import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.functions._
import org.apache.hudi.DataSourceWriteOptions
import org.apache.hudi.DataSourceReadOptions
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.hudi.hive.MultiPartKeysValueExtractor
import org.apache.hudi.hive.HiveSyncConfig
import org.apache.hudi.sync.common.HoodieSyncConfig
```
## 写入 Hudi 数据集
以下示例说明如何创建 DataFrame 并将其写为 Hudi 数据集。
**注意**
要将代码示例粘贴到 Spark shell 中，请在提示符处键入 **:paste**，粘贴示例，然后按 **CTRL** \$1 **D**。
每次向 Hudi 数据集写入时，都必须指定`DataSourceWriteOptions`。 DataFrame 这些选项中的许多选项在写入操作之间可能是相同的。以下示例使用 `hudiOptions` 变量指定常用选项，随后的示例使用这些选项。
### 使用 Amazon EMR 6.7 及更高版本的 Scala 进行写入
**注意**
Amazon EMR 6.7.0 使用 [Apache Hudi 0.11.0-amzn-0](https://hudi.apache.org/)，相比于之前的 Hudi 版本有明显改进。有关更多信息，请参阅 [Apache Hudi 0.11.0 Migration Guide](https://hudi.apache.org/releases/release-0.11.0/#migration-guide)（《Apache Hudi 0.11.0 迁移指南》）。此选项卡上的示例反映了这些更改。
```
// Create a DataFrame
val inputDF = Seq(
("100", "2015-01-01", "2015-01-01T13:51:39.340396Z"),
("101", "2015-01-01", "2015-01-01T12:14:58.597216Z"),
("102", "2015-01-01", "2015-01-01T13:51:40.417052Z"),
("103", "2015-01-01", "2015-01-01T13:51:40.519832Z"),
("104", "2015-01-02", "2015-01-01T12:15:00.512679Z"),
("105", "2015-01-02", "2015-01-01T13:51:42.248818Z")
).toDF("id", "creation_date", "last_update_time")
//Specify common DataSourceWriteOptions in the single hudiOptions variable
val hudiOptions = Map[String,String](
HoodieWriteConfig.TBL_NAME.key -> "tableName",
DataSourceWriteOptions.TABLE_TYPE.key -> "COPY_ON_WRITE",
DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY -> "id",
DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY -> "creation_date",
DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY -> "last_update_time",
DataSourceWriteOptions.HIVE_SYNC_ENABLED_OPT_KEY -> "true",
DataSourceWriteOptions.HIVE_TABLE_OPT_KEY -> "tableName",
DataSourceWriteOptions.HIVE_PARTITION_FIELDS_OPT_KEY -> "creation_date",
HoodieSyncConfig.META_SYNC_PARTITION_EXTRACTOR_CLASS.key -> "org.apache.hudi.hive.MultiPartKeysValueExtractor",
HoodieSyncConfig.META_SYNC_ENABLED.key -> "true",
HiveSyncConfig.HIVE_SYNC_MODE.key -> "hms",
HoodieSyncConfig.META_SYNC_TABLE_NAME.key -> "tableName",
HoodieSyncConfig.META_SYNC_PARTITION_FIELDS.key -> "creation_date"
)
// Write the DataFrame as a Hudi dataset
(inputDF.write
.format("hudi")
.options(hudiOptions)
.option(DataSourceWriteOptions.OPERATION_OPT_KEY,"insert")
.mode(SaveMode.Overwrite)
.save("s3://amzn-s3-demo-bucket/myhudidataset/"))
```
### 使用 Amazon EMR 6.6 及更早版本的 Scala 进行写入
```
// Create a DataFrame
val inputDF = Seq(
("100", "2015-01-01", "2015-01-01T13:51:39.340396Z"),
("101", "2015-01-01", "2015-01-01T12:14:58.597216Z"),
("102", "2015-01-01", "2015-01-01T13:51:40.417052Z"),
("103", "2015-01-01", "2015-01-01T13:51:40.519832Z"),
("104", "2015-01-02", "2015-01-01T12:15:00.512679Z"),
("105", "2015-01-02", "2015-01-01T13:51:42.248818Z")
).toDF("id", "creation_date", "last_update_time")
//Specify common DataSourceWriteOptions in the single hudiOptions variable
val hudiOptions = Map[String,String](
HoodieWriteConfig.TABLE_NAME -> "tableName",
DataSourceWriteOptions.TABLE_TYPE_OPT_KEY -> "COPY_ON_WRITE",
DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY -> "id",
DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY -> "creation_date",
DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY -> "last_update_time",
DataSourceWriteOptions.HIVE_SYNC_ENABLED_OPT_KEY -> "true",
DataSourceWriteOptions.HIVE_TABLE_OPT_KEY -> "tableName",
DataSourceWriteOptions.HIVE_PARTITION_FIELDS_OPT_KEY -> "creation_date",
DataSourceWriteOptions.HIVE_PARTITION_EXTRACTOR_CLASS_OPT_KEY -> classOf[MultiPartKeysValueExtractor].getName
)
// Write the DataFrame as a Hudi dataset
(inputDF.write
.format("org.apache.hudi")
.option(DataSourceWriteOptions.OPERATION_OPT_KEY, DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL)
.options(hudiOptions)
.mode(SaveMode.Overwrite)
.save("s3://amzn-s3-demo-bucket/myhudidataset/"))
```
### 使用写作 PySpark
```
# Create a DataFrame
inputDF = spark.createDataFrame(
[
("100", "2015-01-01", "2015-01-01T13:51:39.340396Z"),
("101", "2015-01-01", "2015-01-01T12:14:58.597216Z"),
("102", "2015-01-01", "2015-01-01T13:51:40.417052Z"),
("103", "2015-01-01", "2015-01-01T13:51:40.519832Z"),
("104", "2015-01-02", "2015-01-01T12:15:00.512679Z"),
("105", "2015-01-02", "2015-01-01T13:51:42.248818Z"),
],
["id", "creation_date", "last_update_time"]
)
# Specify common DataSourceWriteOptions in the single hudiOptions variable
hudiOptions = {
'hoodie.table.name': 'tableName',
'hoodie.datasource.write.recordkey.field': 'id',
'hoodie.datasource.write.partitionpath.field': 'creation_date',
'hoodie.datasource.write.precombine.field': 'last_update_time',
'hoodie.datasource.hive_sync.enable': 'true',
'hoodie.datasource.hive_sync.table': 'tableName',
'hoodie.datasource.hive_sync.partition_fields': 'creation_date',
'hoodie.datasource.hive_sync.partition_extractor_class': 'org.apache.hudi.hive.MultiPartKeysValueExtractor'
}
# Write a DataFrame as a Hudi dataset
inputDF.write \
.format('org.apache.hudi') \
.option('hoodie.datasource.write.operation', 'insert') \
.options(**hudiOptions) \
.mode('overwrite') \
.save('s3://amzn-s3-demo-bucket/myhudidataset/')
```
**注意**
您可能会在代码示例和通知中看到“hoodie”而不是 Hudi。Hudi 代码库广泛使用旧的“hoodie”拼写。
**DataSourceWriteOptions Hudi 的参考资料**
| 选项 | 描述 |
| --- | --- |
| TABLE\$1NAME | 要在其中注册数据集的表名称。 |
| TABLE\$1TYPE\$1OPT\$1KEY | 可选。指定数据集是创建为 `"COPY_ON_WRITE"` 还是 `"MERGE_ON_READ"`。默认值为 `"COPY_ON_WRITE"`。 |
| RECORDKEY\$1FIELD\$1OPT\$1KEY | 其值将用作 `HoodieKey` 的 `recordKey` 组件的记录键字段。实际值将通过对字段值调用 `.toString()` 来获得。可使用点表示法指定嵌套字段，例如 `a.b.c`。 |
| PARTITIONPATH\$1FIELD\$1OPT\$1KEY | 其值将用作 `HoodieKey` 的 `partitionPath` 组件的分区路径字段。实际值将通过对字段值调用 `.toString()` 来获得。 |
| PRECOMBINE\$1FIELD\$1OPT\$1KEY | 在实际写入之前在预合并中使用的字段。如果两个记录具有相同的键值，Hudi 为预合并选择字段值最大的记录（由 `Object.compareTo(..)` 确定）。 |
仅在元数据仓中注册 Hudi 数据集表时才需要以下选项。如果您未将 Hudi 数据集注册为 Hive 元数据仓中的表，则不需要这些选项。
**DataSourceWriteOptions Hive 的参考资料**
| 选项 | 描述 |
| --- | --- |
| HIVE\$1DATABASE\$1OPT\$1KEY | 要同步到的 Hive 数据库。默认值为 `"default"`。 |
| HIVE\$1PARTITION\$1EXTRACTOR\$1CLASS\$1OPT\$1KEY | 用于将分区字段值提取到 Hive 分区列中的类。 |
| HIVE\$1PARTITION\$1FIELDS\$1OPT\$1KEY | 数据集中用于确定 Hive 分区列的字段。 |
| HIVE\$1SYNC\$1ENABLED\$1OPT\$1KEY | 设置为 `"true"` 时，将向 Apache Hive 元数据仓注册数据集。默认值为 `"false"`。 |
| HIVE\$1TABLE\$1OPT\$1KEY | 必需。Hive 中要同步到的表的名称。例如，`"my_hudi_table_cow"`。 |
| HIVE\$1USER\$1OPT\$1KEY | 可选。同步时要使用的 Hive 用户名。例如 `"hadoop"`。 |
| HIVE\$1PASS\$1OPT\$1KEY | 可选。由 `HIVE_USER_OPT_KEY` 指定的用户的 Hive 密码。 |
| HIVE\$1URL\$1OPT\$1KEY | Hive 元数据仓 URL。 |
## 更新插入数据
以下示例演示如何通过编写 a DataFrame 来更新插入数据。与之前的插入示例不同，`OPERATION_OPT_KEY` 值设置为 `UPSERT_OPERATION_OPT_VAL`。此外，还指定 `.mode(SaveMode.Append)` 以指示应追加记录。
### 使用 Amazon EMR 6.7 及更高版本的 Scala 进行更新插入
**注意**
Amazon EMR 6.7.0 使用 [Apache Hudi 0.11.0-amzn-0](https://hudi.apache.org/)，相比于之前的 Hudi 版本有明显改进。有关更多信息，请参阅 [Apache Hudi 0.11.0 Migration Guide](https://hudi.apache.org/releases/release-0.11.0/#migration-guide)（《Apache Hudi 0.11.0 迁移指南》）。此选项卡上的示例反映了这些更改。
```
// Create a new DataFrame from the first row of inputDF with a different creation_date value
val updateDF = inputDF.limit(1).withColumn("creation_date", lit("new_value"))
(updateDF.write
.format("hudi")
.options(hudiOptions)
.option(DataSourceWriteOptions.OPERATION_OPT_KEY, "upsert")
.mode(SaveMode.Append)
.save("s3://amzn-s3-demo-bucket/myhudidataset/"))
```
### 使用 Amazon EMR 6.6 及更早版本的 Scala 进行更新插入
```
// Create a new DataFrame from the first row of inputDF with a different creation_date value
val updateDF = inputDF.limit(1).withColumn("creation_date", lit("new_value"))
(updateDF.write
.format("org.apache.hudi")
.option(DataSourceWriteOptions.OPERATION_OPT_KEY, DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL)
.options(hudiOptions)
.mode(SaveMode.Append)
.save("s3://amzn-s3-demo-bucket/myhudidataset/"))
```
### 使用 Upsert PySpark
```
from pyspark.sql.functions import lit
# Create a new DataFrame from the first row of inputDF with a different creation_date value
updateDF = inputDF.limit(1).withColumn('creation_date', lit('new_value'))
updateDF.write \
.format('org.apache.hudi') \
.option('hoodie.datasource.write.operation', 'upsert') \
.options(**hudiOptions) \
.mode('append') \
.save('s3://amzn-s3-demo-bucket/myhudidataset/')
```
## 删除记录
要硬删除记录，您可以更新插入一个空的负载。在这种情况下，`PAYLOAD_CLASS_OPT_KEY` 选项指定 `EmptyHoodieRecordPayload` 类。该示例使用 upsert 示例中使用的相同方法来指定相同的记录。 DataFrame `updateDF`
### 使用 Amazon EMR 6.7 及更高版本的 Scala 进行删除
**注意**
Amazon EMR 6.7.0 使用 [Apache Hudi 0.11.0-amzn-0](https://hudi.apache.org/)，相比于之前的 Hudi 版本有明显改进。有关更多信息，请参阅 [Apache Hudi 0.11.0 Migration Guide](https://hudi.apache.org/releases/release-0.11.0/#migration-guide)（《Apache Hudi 0.11.0 迁移指南》）。此选项卡上的示例反映了这些更改。
```
(updateDF.write
.format("hudi")
.options(hudiOptions)
.option(DataSourceWriteOptions.OPERATION_OPT_KEY, "delete")
.mode(SaveMode.Append)
.save("s3://amzn-s3-demo-bucket/myhudidataset/"))
```
### 使用 Amazon EMR 6.6 及更早版本的 Scala 进行删除
```
(updateDF.write
.format("org.apache.hudi")
.option(DataSourceWriteOptions.OPERATION_OPT_KEY, DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL)
.option(DataSourceWriteOptions.PAYLOAD_CLASS_OPT_KEY, "org.apache.hudi.common.model.EmptyHoodieRecordPayload")
.mode(SaveMode.Append)
.save("s3://amzn-s3-demo-bucket/myhudidataset/"))
```
### 使用删除 PySpark
```
updateDF.write \
.format('org.apache.hudi') \
.option('hoodie.datasource.write.operation', 'upsert') \
.option('hoodie.datasource.write.payload.class', 'org.apache.hudi.common.model.EmptyHoodieRecordPayload') \
.options(**hudiOptions) \
.mode('append') \
.save('s3://amzn-s3-demo-bucket/myhudidataset/')
```
您还可以通过以下方式硬删除数据：将 `OPERATION_OPT_KEY ` 设置为 `DELETE_OPERATION_OPT_VAL` 来删除您提交的数据集中的所有记录。有关执行软删除的说明，以及有关删除 Hudi 表中存储的数据的详细信息，请参阅 Apache Hudi 文档中的 [Deletes](https://hudi.apache.org/docs/writing_data.html#deletes)。
## 从 Hudi 数据集读取
要在当前时间点检索数据，Hudi 默认情况下执行快照查询。以下是查询在 [写入 Hudi 数据集](#emr-hudi-dataframe) 中写入 S3 的数据集的示例。*s3://amzn-s3-demo-bucket/myhudidataset*替换为表路径，为每个分区级别添加通配符星号，*再加上一个*星号。在此示例中，有一个分区级别，因此我们添加了两个通配符号。
### 使用 Amazon EMR 6.7 及更高版本的 Scala 进行读取
**注意**
Amazon EMR 6.7.0 使用 [Apache Hudi 0.11.0-amzn-0](https://hudi.apache.org/)，相比于之前的 Hudi 版本有明显改进。有关更多信息，请参阅 [Apache Hudi 0.11.0 Migration Guide](https://hudi.apache.org/releases/release-0.11.0/#migration-guide)（《Apache Hudi 0.11.0 迁移指南》）。此选项卡上的示例反映了这些更改。
```
val snapshotQueryDF = spark.read
.format("hudi")
.load("s3://amzn-s3-demo-bucket/myhudidataset")
.show()
```
### 使用 Amazon EMR 6.6 及更早版本的 Scala 进行读取
```
(val snapshotQueryDF = spark.read
.format("org.apache.hudi")
.load("s3://amzn-s3-demo-bucket/myhudidataset" + "/*/*"))
snapshotQueryDF.show()
```
### 使用 “阅读” PySpark
```
snapshotQueryDF = spark.read \
.format('org.apache.hudi') \
.load('s3://amzn-s3-demo-bucket/myhudidataset' + '/*/*')
snapshotQueryDF.show()
```
### 递增查询
您还可以使用 Hudi 执行增量查询，以获取自给定提交时间戳以来已更改的记录流。为此，请将 `QUERY_TYPE_OPT_KEY` 字段设置为 `QUERY_TYPE_INCREMENTAL_OPT_VAL`。然后，为 `BEGIN_INSTANTTIME_OPT_KEY` 添加一个值，以获取自指定时间以来写入的所有记录。递增查询的效率通常是批处理查询的十倍，因为它们只处理更改的记录。
执行增量查询时，请使用根（基）表路径，而不需要用于快照查询的通配符星号。
**注意**
Presto 不支持递增查询。
#### 使用 Scala 进行增量查询
```
val incQueryDF = spark.read
.format("org.apache.hudi")
.option(DataSourceReadOptions.QUERY_TYPE_OPT_KEY, DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL)
.option(DataSourceReadOptions.BEGIN_INSTANTTIME_OPT_KEY, )
.load("s3://amzn-s3-demo-bucket/myhudidataset")
incQueryDF.show()
```
#### 使用增量查询 PySpark
```
readOptions = {
'hoodie.datasource.query.type': 'incremental',
'hoodie.datasource.read.begin.instanttime': ,
}
incQueryDF = spark.read \
.format('org.apache.hudi') \
.options(**readOptions) \
.load('s3://amzn-s3-demo-bucket/myhudidataset')
incQueryDF.show()
```
有关从 Hudi 数据集读取的更多信息，请参阅 Apache Hudi 文档中的 [查询 Hudi 表](https://hudi.apache.org/docs/querying_data.html)。
# 使用 Hudi CLI
您可以使用 Hudi CLI 管理 Hudi 数据集，以查看有关提交、文件系统、统计信息等的信息。还可以使用 CLI 手动执行压缩、计划压缩或取消计划的压缩。有关更多信息，请参阅 Apache Hudi 文档中的[ CLI 互动](https://hudi.apache.org/docs/cli/)。
**启动 Hudi CLI 并连接到数据集**
1. 使用 SSH 连接主节点。有关更多信息，请参阅《Amazon EMR 管理指南》**中的[使用 SSH 连接到主节点](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-connect-master-node-ssh.html)。
1. 在命令行中，键入 `/usr/lib/hudi/cli/bin/hudi-cli.sh`。
命令提示符更改为 `hudi->`。
1. 使用以下命令连接到数据集。*s3://amzn-s3-demo-bucket/myhudidataset*替换为要使用的数据集的路径。我们使用的值与前面示例中建立的值相同。
```
connect --path s3://amzn-s3-demo-bucket/myhudidataset
```
命令提示符将更改以包括您连接到的数据集，如以下示例所示。
```
hudi:myhudidataset->
```
默认情况下，Amazon EMR 版本 7.3.0 至 7.8.0 中的 `hudi-cli.sh` 脚本使用 `hudi-cli-bundle.jar`。如果遇到问题，可以使用以下命令切换回经典 Hudi CLI：
```
/usr/lib/hudi/cli/bin/hudi-cli.sh --cliBundle false
```
此命令运行 `hudi-cli.sh` 脚本，设置 `--cliBundle` 标志，并指示 CLI 使用单个 JAR 文件而不是捆绑的 JAR。默认情况下，将 `--cliBundle` 设置为 true，这意味着 CLI 改用捆绑的 JAR。
## 使用 Amazon EMR 7.9.0 及更高版本
**注意**
在 EMR 7.9.0 及更高版本中，**hudi-cli.sh** 脚本已被弃用。**亚马逊 EMR 7.9.0 及更高版本使用.jar。hudi-cli-bundle**
**要启动 Hudi CLI 并连接到数据集，请执行以下操作：**
1. 使用 SSH 连接主节点。有关更多信息，请参阅《Amazon EMR 管理指南》**中的[使用 SSH 连接到主节点](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-connect-master-node-ssh.html)。
1. 在命令行中，键入**/usr/lib/hudi/cli-bundle/bin/hudi-cli-with-bundle .sh** 或者直接键入**hudi-cli-with-bundle**或 **>** hudi-cli。
命令提示符更改为 **hudi- >**。
1. 使用以下命令连接到数据集。将 **s3://amzn-s3-demo-bucket/myhudidataset** 替换为您想要使用的数据集的路径。我们使用的值与前面示例中建立的值相同。
```
connect --path s3://amzn-s3-demo-bucket/myhudidataset
```
1. 命令提示符将更改以包括您连接到的数据集，如以下示例所示。
```
hudi:myhudidataset->
```
# Hudi 发行版历史记录
下表列出了 Amazon EMR 每个发行版中包含的 Hudi 版本，以及随应用程序一起安装的组件。有关每个发行版本中的组件版本，请参阅 [Amazon EMR 7.x 发行版](emr-release-7x.md)、[Amazon EMR 6.x 发行版](emr-release-6x.md) 或 [Amazon EMR 5.x 发行版](emr-release-5x.md) 中的发行版“组件版本”部分。
**Hudi 版本信息。**
| Amazon EMR 发行版标签 | Hudi 版本 | 随 Hudi 安装的组件 |
| --- | --- | --- |
| emr-7.12.0 | 1.0.2-amzn-1 | Not available. |
| emr-7.11.0 | 1.0.2-amzn-0 | Not available. |
| emr-7.10.0 | 0.15.0-amzn-7 | Not available. |
| emr-7.9.0 | 0.15.0-amzn-6 | Not available. |
| emr-7.8.0 | 0.15.0-amzn-5 | Not available. |
| emr-7.7.0 | 0.15.0-amzn-4 | Not available. |
| emr-7.6.0 | 0.15.0-amzn-3 | Not available. |
| emr-7.5.0 | 0.15.0-amzn-2 | Not available. |
| emr-7.4.0 | 0.15.0-amzn-1 | Not available. |
| emr-7.3.0 | 0.15.0-amzn-0 | Not available. |
| emr-7.2.0 | 0.14.1-amzn-1 | Not available. |
| emr-5.36.2 | 0.10.1-amzn-1 | Not available. |
| emr-7.1.0 | 0.14.1-amzn-0 | Not available. |
| emr-7.0.0 | 0.14.0-amzn-1 | Not available. |
| emr-6.15.0 | 0.14.0-amzn-0 | Not available. |
| emr-6.14.0 | 0.13.1-amzn-2 | Not available. |
| emr-6.13.0 | 0.13.1-amzn-1 | Not available. |
| emr-6.12.0 | 0.13.1-amzn-0 | Not available. |
| emr-6.11.1 | 0.13.0-amzn-0 | Not available. |
| emr-6.11.0 | 0.13.0-amzn-0 | Not available. |
| emr-6.10.1 | 0.12.2-amzn-0 | Not available. |
| emr-6.10.0 | 0.12.2-amzn-0 | Not available. |
| emr-6.9.1 | 0.12.1-amzn-0 | Not available. |
| emr-6.9.0 | 0.12.1-amzn-0 | Not available. |
| emr-6.8.1 | 0.11.1-amzn-0 | Not available. |
| emr-6.8.0 | 0.11.1-amzn-0 | Not available. |
| emr-6.7.0 | 0.11.0-amzn-0 | Not available. |
| emr-5.36.1 | 0.10.1-amzn-1 | Not available. |
| emr-5.36.0 | 0.10.1-amzn-1 | Not available. |
| emr-6.6.0 | 0.10.1-amzn-0 | Not available. |
| emr-5.35.0 | 0.9.0-amzn-2 | Not available. |
| emr-6.5.0 | 0.9.0-amzn-1 | Not available. |
| emr-6.4.0 | 0.8.0-amzn-0 | Not available. |
| emr-6.3.1 | 0.7.0-amzn-0 | Not available. |
| emr-6.3.0 | 0.7.0-amzn-0 | Not available. |
| emr-6.2.1 | 0.6.0-amzn-1 | Not available. |
| emr-6.2.0 | 0.6.0-amzn-1 | Not available. |
| emr-6.1.1 | 0.5.2-incubating-amzn-2 | Not available. |
| emr-6.1.0 | 0.5.2-incubating-amzn-2 | Not available. |
| emr-6.0.1 | 0.5.0-incubating-amzn-1 | Not available. |
| emr-6.0.0 | 0.5.0-incubating-amzn-1 | Not available. |
| emr-5.34.0 | 0.9.0-amzn-0 | Not available. |
| emr-5.33.1 | 0.7.0-amzn-1 | Not available. |
| emr-5.33.0 | 0.7.0-amzn-1 | Not available. |
| emr-5.32.1 | 0.6.0-amzn-0 | Not available. |
| emr-5.32.0 | 0.6.0-amzn-0 | Not available. |
| emr-5.31.1 | 0.6.0-amzn-0 | Not available. |
| emr-5.31.0 | 0.6.0-amzn-0 | Not available. |
| emr-5.30.2 | 0.5.2-incubating | Not available. |
| emr-5.30.1 | 0.5.2-incubating | Not available. |
| emr-5.30.0 | 0.5.2-incubating | Not available. |
| emr-5.29.0 | 0.5.0-incubating | Not available. |
| emr-5.28.1 | 0.5.0-incubating | Not available. |
| emr-5.28.0 | 0.5.0-incubating | Not available. |