Amazon EMR 上 Apache Hudi 0.9.0 提供的新功能

Apache Hudi 是一个开源事务性数据湖框架，极大地简化了增量数据处理和数据管道开发。它面向 Amazon Simple Storage Service（Amazon S3）或 Apache HDFS 上的数据湖，提供了事务支持和记录级的插入、更新和删除功能，从而实现了简化。Apache Hudi 支持与 Apache Spark、Apache Hive、Presto 和 Trino 等开源大数据分析框架集成。此外，Apache Hudi 还允许您以开放格式（例如 Apache Parquet 和 Apache Avro），在 Amazon S3 或 Apache HDFS 中维护数据。

客户使用 Apache Hudi 的常见使用场景如下：

简化数据摄入管道，这些管道用于处理来自流式传输和批量数据源的延迟到达或更新记录。
使用更改数据捕获（CDC, Change Data Capture）从事务系统中摄入数据。
实施数据删除管道以遵守数据隐私法规，例如 GDPR（General Data Protection Regulation, 通用数据保护条例）合规性。遵守 GDPR 是当今现代化数据架构的必要条件，这包括“擦除权”或“被遗忘权”功能，并且可以使用 Apache Hudi 功能来实施，用于替代删除和更新操作。

我们很高兴地宣布，Apache Hudi 0.9.0 已在 Amazon EMR 5.34 和 EMR 6.5.0 上提供。这是一个主要版本，其亮点在于包括了 Spark SQL DML 和 DDL 支持，以及其他一些写入器/读取器方面的改进。与 Hudi 0.6.0 相比，我们观察到的查询性能提高了 3 倍，这尤为显著，因此，如果您希望实施事务数据湖来达到极高的更新插入和删除操作速度，或者正在使用旧版本的Hudi，那么这是一个极佳的版本。在这篇文章中，我们将重点介绍 0.9.0 版本提供的以下新功能和改进：

Spark SQL DML 和 DDL 支持：探索 Spark SQL DML 和 DDL 支持。
性能改进：探索写入器和查询端引入的性能改进和与性能相关的新功能。
其他功能：探索其他有用的功能，例如基于 Amazon DynamoDB 的 Optimistic Concurrency Control（OCC）锁定、删除分区操作等。

Spark SQL DML 和 DDL 支持

最令人兴奋的新功能是 Apache Hudi 0.9.0 增加了对使用 Spark SQL 的 DDL/DML 的支持。这使得 Hudi 面向所有人（非工程师、分析师等）提供更好的可访问性和可操作性迈出了重要一步。此外，它还使现有数据集可以轻松地迁移到 Apache Hudi 表中，并且它更接近使用 Spark SQL DML 和 DDL 的低代码范例，因此无需编写 scala/python 代码。

用户现在可以使用 CREATE TABLE....USING HUDI 和 CREATE TABLE ..AS SELECT SQL 语句创建表，以直接在 AWS Glue 目录中管理表。

然后，用户可以使用 INSERT、UPDATE、MERGE INTO 和 DELETE SQL 语句操作数据。INSERT OVERWRITE 语句可用于针对现有批处理 ETL 管道中的表或分区，覆盖其现有数据。

我们来看一个简单的示例，在该示例中，我们创建了一个 Hudi 表 amazon_customer_review_hudi，模仿了 Amazon Customer Reviews 公有数据集并执行以下活动：

先决条件：创建 Amazon Simple Storage Service（S3）存储桶 s3://EXAMPLE-BUCKET 和 s3://EXAMPLE-BUCKET-1
创建分区的 Hudi 表amazon_product_review_hudi
创建源 Hudi 表 amazon_customer_review_parquet_merge_source，其内容将与 amazon_product_review_hudi 表合并
将数据插入 amazon_customer_review_parquet_merge_source 和 amazon_product_review_hudi 中，以及通过读取
amazon_customer_review_parquet_merge_source 中的数据并与 Hudi 表 amazon_product_review_hudi 合并来执行合并操作
在 amazon_customer_review_hudi 上对之前插入的记录执行删除操作

配置 Spark 会话

我们通过 EMR studio 笔记本使用以下脚本，用于配置 Spark 会话，以便结合使用 Apache Hudi DML 和 DDL 支持。以下示例演示了如何启动交互式 Spark shell、使用 Spark 提交或将 Amazon EMR Notebooks 与 Amazon EMR 上的 Hudi 结合使用。我们建议使用以下 Apache Livy 配置启动 EMR 集群：

[
    {
        "Classification": "livy-conf",
        "Properties": {
            "livy.file.local-dir-whitelist": "/usr/lib/hudi"
        }
    }
]

上述配置让您可以在配置 Spark 会话时，直接引用 EMR leader 节点上的本地 /usr/lib/hudi/hudi-spark-bundle.jar。或者，您也可以将 /usr/lib/hudi/hudi-spark-bundle.jar 复制到 HDFS 位置，并在初始化 Spark 会话时引用该项。以下是从笔记本初始化 Spark 会话的命令：

%%configure -f
{
    "conf" : {
        "spark.jars":"file:///usr/lib/hudi/hudi-spark-bundle.jar",
        "spark.serializer":"org.apache.spark.serializer.KryoSerializer",
        "spark.sql.extensions":"org.apache.spark.sql.hudi.HoodieSparkSessionExtension"
    }
}

创建表

我们创建以下 Apache Hudi 表：amazon_customer_review_hudi 和 amazon_customer_review_parquet_merge_source

amazon_customer_review_hudi and amazon_customer_review_parquet_merge_source

%%sql 

/****************************
创建 HUDI 表，该表与包含所选列的 Amazon Customer Reviews 表具有相同架构 
*****************************/

-- Hudi 0.9.0 配置 https://hudi.apache.org/docs/configurations
-- Hudi 配置可以在选项块中设置为 hoodie.datasource.hive_sync.assume_date_partitioning = 'false'，


create table if not exists amazon_customer_review_hudi
    ( marketplace string,
      review_id string,
      customer_id string,
      product_title string,
      star_rating int,
      timestamp long ,
      review_date date,
      year string,
      month string ,
      day string
      )
      using hudi
      location 's3://EXAMPLE-BUCKET/my-hudi-dataset/'
      options ( 
      type = 'cow', 
      primaryKey = 'review_id',
      preCombineField = 'timestamp',
      hoodie.datasource.write.hive_style_partitioning = 'true'
      )
      partitioned by (year,month,day);
      

-- 将位置 's3://EXAMPLE-BUCKET/my-hudi-dataset/' 更改为您已在 AWS 账户中创建的对应 S3 存储桶

%%sql 
/****************************
创建用作源的 amazon_customer_review_parquet_merge_source，以合并到 amazon_customer_review_hudi。
该表包含 deleteRecord 列，用于跟踪是否需要删除记录
*****************************/


create table if not exists amazon_customer_review_parquet_merge_source 
       (
        marketplace string,
        review_id string,
        customer_id string,
        product_title string,
        star_rating int,
        review_date date,
        deleteRecord string
       )
       STORED AS PARQUET
       LOCATION 's3://EXAMPLE-BUCKET-1/toBeMergeData/'


-- 将位置 (s3://EXAMPLE-BUCKET-1/toBeMergeData/') 更改为您已在 AWS 账户中创建的对应 S3 存储桶

为了进行比较，如果使用编程方法创建 amazon_customer_review_hudi，则 PySpark 示例代码如下所示。

# 创建 DataFrame
inputDF = spark.createDataFrame(
    [
         ("Italy", "11", "1111", "table", 5, 1648126827, "2015/05/02", "2015", "05", "02"),
         ("Spain", "22", "2222", "chair", 5, 1648126827, "2015/05/02", "2015", "05", "02")        
    ],
    ["marketplace", "review_id", "customer_id", "product_title", "star_rating", "timestamp", "review_date", "year", "month", "day" ]
)

# 输出 inputDF 的架构 
inputDF.printSchema()

# 在单个 hudiOptions 变量中指定通用的 DataSourceWriteOptions
hudiOptions = {
"hoodie.table.name": "amazon_customer_review_hudi",
"hoodie.datasource.write.recordkey.field": "review_id",
"hoodie.datasource.write.partitionpath.field": "year,month,day",
"hoodie.datasource.write.precombine.field": "timestamp",
"hoodie.datasource.write.hive_style_partitioning": "true",
"hoodie.datasource.hive_sync.enable": "true",
"hoodie.datasource.hive_sync.table": " amazon_customer_review_hudi",
"hoodie.datasource.hive_sync.partition_fields": "year,month,day",
"hoodie.datasource.hive_sync.partition_extractor_class": "org.apache.hudi.hive.MultiPartKeysValueExtractor"
}


# 创建 Hudi 表并将数据插入位于所指定 S3 位置的 my_hudi_table_1 hudi 表中 
inputDF.write \
       .format("org.apache.hudi")\
       .option("hoodie.datasource.write.operation", "insert")\
       .options(**hudiOptions)\
       .mode("append")\
       .save("s3://EXAMPLE-BUCKET/my-hudi-dataset/")

将数据插入 Hudi 表

我们将记录插入表 amazon_customer_review_parquet_merge_source 以供合并操作使用。这包括插入一行以刷新插入、更新和删除。

%%sql 

/****************************
 将记录插入 amazon_customer_review_parquet_merge_source 以进行删除 
*****************************/

-- 将 merge as deleteRecord 设置为 yes 后，将从 amazon_customer_review_hudi 中删除记录

insert into amazon_customer_review_parquet_merge_source
    select
    'italy',
    '11',
    '1111',
    'table',
     5,
    TO_DATE(CAST(UNIX_TIMESTAMP('2015/05/02', 'yyyy/MM/dd') AS TIMESTAMP)) as  review_date,
    'yes' 
    
   

%%sql
/****************************
 将记录插入 amazon_customer_review_parquet_merge_source 以进行更新
*****************************/

-- 在合并后，将从 amazon_customer_review_hudi 使用新的 Star rating 和 product_title 更新记录

insert into amazon_customer_review_parquet_merge_source
    select
    'spain',
    '22',
    '2222',
    'Relaxing chair',
     4,
    TO_DATE(CAST(UNIX_TIMESTAMP('2015/05/02', 'yyyy/MM/dd') AS TIMESTAMP)) as  review_date,
    'no' 


%%sql
/****************************
 将记录插入 amazon_customer_review_parquet_merge_source 以进行插入 
*****************************/

-- 合并后，记录将插入 amazon_customer_review_hudi 

insert into amazon_customer_review_parquet_merge_source
    select
    'uk',
    '33',
    '3333',
    'hanger',
     3,
    TO_DATE(CAST(UNIX_TIMESTAMP('2015/05/02', 'yyyy/MM/dd') AS TIMESTAMP)) as  review_date,
    'no'

现在，我们将记录插入用作合并操作目标表的 amazon_customer_review_hudi 表。

%%sql

/****************************
 在合并后，将记录插入 amazon_customer_review_hudi 表以进行删除 
*****************************/

-- Spark SQL 日期时间函数 https://spark.apache.org/docs/latest/api/sql/index.html#date_add

insert into amazon_customer_review_hudi 
    select 
    'italy',
    '11',
    '1111',
    'table',
     5,
    unix_timestamp(current_timestamp()) as timestamp,
    TO_DATE(CAST(UNIX_TIMESTAMP('2015/05/02', 'yyyy/MM/dd') AS TIMESTAMP)) as  review_date,
    date_format(date '2015-05-02', "yyyy") as year,
    date_format(date '2015-05-02', "MM") as month,
    date_format(date '2015-05-02', "dd") as day  


%%sql
/****************************
 在合并后，将记录插入 amazon_customer_review_hudi table 以进行更新 
*****************************/

insert into  amazon_customer_review_hudi
    select 
    'spain',
    '22',
    '2222',
    'chair ',
     5,
    unix_timestamp(current_timestamp()) as timestamp,
    TO_DATE(CAST(UNIX_TIMESTAMP('2015/05/02', 'yyyy/MM/dd') AS TIMESTAMP)) as  review_date,
    date_format(date '2015-05-02', "yyyy") as year,
    date_format(date '2015-05-02', "MM") as month,
    date_format(date '2015-05-02', "dd") as day

合并到

我们来执行从 amazon_customer_review_parquet_merge_source 合并到 amazon_customer_review_hudi。

%%sql 

/*************************************
MergeInto：将源合并到目标 
**************************************/

-- 源 amazon_customer_review_parquet_merge_source 
-- 目标 amazon_customer_review_hudi

merge into amazon_customer_review_hudi as target
using ( 
        select
        marketplace,
        review_id,
        customer_id,
        product_title,
        star_rating,
        review_date,
        deleteRecord,
        date_format(review_date, "yyyy") as year,
        date_format(review_date, "MM") as month,
        date_format(review_date, "dd") as day
        from amazon_customer_review_parquet_merge_source ) source
on target.review_id = source.review_id 
when matched and deleteRecord != 'yes' then 

update set target.timestamp = unix_timestamp(current_timestamp()), 
target.star_rating = source.star_rating,
target.product_title = source.product_title

when matched and deleteRecord = 'yes' then delete

when not matched then insert 
      ( target.marketplace,
        target.review_id,
        target.customer_id,
        target.product_title,
        target.star_rating,
        target.timestamp ,
        target.review_date,
        target.year ,
        target.month  ,
        target.day
      ) 
      values
      (
        source.marketplace,
        source.review_id,
        source.customer_id,
        source.product_title,
        source.star_rating,
        unix_timestamp(current_timestamp()),
        source.review_date,
        source.year ,
        source.month ,
        source.day 
       )

注意事项和限制

到目前为止，合并条件只能应用于主键。
-- 合并条件只能用于主键
on target.review_id = source.review_id
写入时复制（CoW, Copy on Write）表上支持部分更新，但读取时合并（MoR, Merge on Read）表上不支持。
目标表的字段不能是 MoR 表的更新表达式的右侧值：
-- 更新将导致错误，因为目标列位于表达式的右侧
update set target.star_rating = target.star_rating +1

删除记录

现在我们删除插入的记录。

%%sql

/*************************************
从 amazon_customer_review_hudi 表中删除插入的记录 
**************************************/
Delete from amazon_customer_review_hudi where review_id == '22'


%%sql 
/*************************************
从 amazon_customer_review_hudi 表中查询已删除的记录 
**************************************/
select * from amazon_customer_review_hudi where review_id == '22'

架构演变

Hudi 支持常见的模式演变场景，例如添加可为空的字段或提升字段的数据类型。我们将新列 ssid（类型为 int）添加到现有 amazon_customer_review_hudi 表中，并使用额外的列插入记录。Hudi 允许查询具有更新的表架构的新旧数据和新数据。

%%sql

/*************************************
将类型为 int 的新列名 ssid 添加到 amazon_customer_review_hudi 表
**************************************/

ALTER TABLE amazon_customer_review_hudi ADD COLUMNS (ssid int)

%%sql
/*************************************
将新记录添加到更改后的表 amazon_customer_review_hudi 
**************************************/
insert into amazon_customer_review_hudi
    select 
    'germany',
    '55',
    '5555',
    'car',
     5,
    unix_timestamp(current_timestamp()) as timestamp,
    TO_DATE(CAST(UNIX_TIMESTAMP('2015/05/02', 'yyyy/MM/dd') AS TIMESTAMP)) as  review_date,
    10 as ssid,
    date_format(date '2015-05-02', "yyyy") as year,
    date_format(date '2015-05-02', "MM") as month,
    date_format(date '2015-05-02', "dd") as day  

%%sql 
/*************************************
将 ssid 类型从 int 提升为 long  
**************************************/
ALTER TABLE amazon_customer_review_hudi CHANGE COLUMN ssid ssid long


%%sql 
/*************************************
从 amazon_customer_review_hudi 表中查询数据
**************************************/
select * from amazon_customer_review_hudi where review_id == '55'

Spark 性能改进

查询端改进

Apache Hudi 表现在已在元存储中注册为 Spark 数据源表。这使得 Hudi 表上的 Spark SQL 查询能够针对在写入时复制的表使用 Spark 的原生 Parquet Reader，针对读取时合并的表使用 Hudi 的自定义 MergeOnReadSnapshotRelation。因此，它不再依赖于 Spark 中的 Hive 输入格式回退，后者的维护和效率不如 Spark 的原生读取器。这带来了许多优化，例如使用 Spark 的原生 Parquet Reader，以及实施 Hudi 自己的 Spark FileIndex 实施。文件索引通过优化缓存、支持分区修剪以及通过 Hudi 元数据表（而不是直接从 Amazon S3 列出）列出文件的能力，帮助提高文件列表性能。此外，Hudi 现在支持通过 Spark 数据源进行时间旅行查询，这使您可以查询历史时间点的数据集快照。

其他需要注意的重要事项包括：

在通过 Spark SQL 进行查询时，不再需要 spark.sql.hive.convertMetastoreParquet=false 和 mapreduce.input.pathFilter.class=org.apache.hudi.hadoop.HoodieROTablePathFilter 这样的配置。
现在，通过数据源 API 查询 Hudi 数据集时，您可以使用非全局查询路径。这使您可以通过基本路径查询表，而不必在查询路径中指定 *。

我们运行了一个源于 3 TB 规模 TPC-DS 基准测试的性能基准测试，以确定对于写入时复制的表，EMR 6.5.0 上的 Hudi 0.9.0 相对于 EMR 6.2.0 上的 Hudi 0.6.0（2021 年初）的查询性能提升。这些查询运行在 5 节点的 c5.9xlarge EMR 集群中。

从几何均值来看，使用 Hudi 0.9.0 的查询比使用 Hudi 0.6.0 的查询要快 3 倍。下图比较了对于两个 Amazon EMR/Hudi 版本，在 TPC-DS 3TB 查询数据集中运行全部查询的总累计运行时和运行时几何均值（值越低越好）。

从几何均值来看，使用 Hudi 0.9.0 的查询比使用 Hudi 0.6.0 的查询要快 3 倍。

写入器方面的改进

虚拟键支持

Apache Hudi 通过向数据集添加额外的列来维护元数据。这使它能够支持 upsert/delete 操作以及与之相关的各种功能，例如增量查询、压缩等。这些元数据列（即 _hoodie_commit_time、_hoodie_record_key、_hoodie_partition_path、_hoodie_file_name 和 _hoodie_commit_seqno）让 Hudi 可以唯一地标识记录、记录所在的分区/文件以及更新了记录的最后一次提交。

但是，生成和维护这些元数据列会增加 Hudi 表在磁盘上占用的存储空间。其中一些列，例如 _hoodie_record_key 和 _hoodie_partition_path，可以从已存储在数据集中的其他数据列构造。Apache Hudi 0.9.0 引入了对虚拟键的支持。这使得用户可以禁用这些元数据列的生成，改为依靠实际的数据列，使用相应的键生成器动态构造记录键/分区路径。这有助于减少存储空间占用，以及缩短摄取入时间。但是，此功能需要注意以下几点：

它只可用于仅附加/不可改变数据。它不能用于需要更新插入和删除的使用场景，这些情况下需要 _hoodie_record_key 和 _hoodie_partition_path 等列，以使 Bloom 索引正常工作。
不支持增量查询，因为它们需要 _hoodie_commit_time 筛选在特定时间写入/更新的记录。
启用此功能后，将无法为现有表关闭该功能。

默认情况下，该功能处于关闭状态，可以通过将 hoodie.populate.meta.fields 设置为 false 来启用。我们对公开的 Amazon Customer Reviews 数据集，使用批量插入衡量了写入性能和存储占用空间的改进。以下是我们使用的代码片段：

import org.apache.hudi.DataSourceWriteOptions
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.spark.sql.SaveMode

var srcPath = "s3://amazon-reviews-pds/parquet/"
var tableName = "amazon_reviews_table"
var tablePath = "s3://<bucket>/<prefix>/" + tableName

val inputDF = spark.read.format("parquet").load(srcPath)

inputDF.write.format("hudi")
 .option(HoodieWriteConfig.TABLE_NAME, tableName)
 .option(DataSourceWriteOptions.OPERATION_OPT_KEY, DataSourceWriteOptions.BULK_INSERT_OPERATION_OPT_VAL)
 .option(DataSourceWriteOptions.TABLE_TYPE_OPT_KEY, DataSourceWriteOptions.COW_TABLE_TYPE_OPT_VAL)
 .option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "review_id")
 .option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY, "product_category") 
 .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "review_date")
 .option("hoodie.populate.meta.fields", "<true/false>")
 .mode(SaveMode.Overwrite)
 .save(tablePath)

该实验在四节点 c4.2xlarge EMR 集群（一个 leader，三个 core）上运行。我们观察到，启用虚拟键后，写入运行时性能提高 10.63%，存储空间占用减少 8.67%。下图比较了使用和不使用虚拟键的批量插入运行时和表大小（值越小越好）：

基于时间轴服务器的标记机制

Apache Hudi 在写入操作期间，支持自动清理写入的未提交数据。通过生成与每个数据文件对应的标记文件来支持这种清理，利用这种方法可以跟踪感兴趣的数据文件，而不必通过列出所有文件来扫描整个表。尽管相比在整个表中扫描未提交的数据文件，现有标记机制的效率要高得多，但它仍然会对 Amazon S3 数据湖产生性能影响。例如，写入大量标记文件（每个数据文件一个标记），然后在成功提交后将其删除可能会需要相当长的时间，有时大约需要几分钟。此外，同时写入大量数据/标记文件时，可能会达到 Amazon S3 节流限制。

Apache Hudi 0.9.0 为这种标记机制引入了基于时间轴服务器的新实施。这提高了整体写入性能，并显著降低达到 Amazon S3 节流限制的可能性，从而提高了处理 Amazon S3 工作负载的效率。新机制将 Hudi 的时间轴服务器组件作为一个中心位置，用于处理（来自所有执行程序的）所有标记创建/删除请求，从而实现了对这些请求的批量处理并减少向 Amazon S3 发出的请求数量。因此，拥有 Amazon S3 数据湖的用户可以利用此功能来提高写操作性能，并避免因标记文件管理而导致的节流。这对于写入大量数据文件（例如 1 万或更多）的场景的影响尤为明显。

这种新的机制默认情况下不启用，可以通过为写操作将 hoodie.write.markers.type 设置为 timeline_server_based 来启用。有关该功能的详细信息，请参阅 Apache Hudi 社区的此博文。

其他改进

基于 DynamoDB 的锁定

Optimistic Concurrency Control 是 Apache Hudi 0.8.0 引入的主要功能之一，它允许多个并发写入器将数据提取到同一个 Hudi 表中。该功能需要获取锁，为此您可以使用 Zookeeper（原定设置为在 EMR 上）或 HiveMetaStore。但是，这些锁提供程序要求所有写入器与 Zookeeper/Hive 元存储运行在同一个集群上。

Amazon EMR 上的 Apache Hudi 0.9.0 引入了 DynamoDB 作为锁提供程序。这允许跨不同集群运行的多个写入器将数据提取到同一个 Hudi 表中。此功能最初是在 Amazon EMR 上的 Hudi 0.9.0 中添加的，并重新用于开源 Hudi 的 0.10.0 版本中。要对此进行配置，应设置以下属性：

配置	值	描述	必填
hoodie.write.lock.provider	`org.apache.hudi.client. transaction.lock. DynamoDBBasedLockProvider`	所使用的锁提供程序实施	是
hoodie.write.lock.dynamodb. table	<字符串>	用于获取锁定的 DynamoDB 表名称。如果该表不存在，则会创建该表。同一个表可在您的处理相同或不同表的所有 Hudi 作业上使用	是
hoodie.write.lock.dynamodb. partition_key	<字符串>	用于锁定表分区键属性的字符串值。它必须是唯一标识 Hudi 表的字符串，例如 Hudi 表名	否。原定设置：`Hudi Table Name`
hoodie.write.lock.dynamodb. region	<字符串>	DynamoDB 锁定表所在的 AWS 区域或者必须在其中创建该表的区域。	否。原定设置： `us-east-1`
hoodie.write.lock.dynamodb. billing_mode	<字符串>	创建锁定表时使用的 DynamoDB 计费模式。如果表已经存在，则此项不起作用	否。原定设置： `PAY_PER_REQUEST`
hoodie.write.lock.dynamodb. read_capacity	<整数>	创建锁定表时使用的 DynamoDB 读取容量。如果表已经存在，则此项不起作用	否。原定设置：`20`
hoodie.write.lock.dynamodb. write_capacity	<整数>	创建锁定表时使用的 DynamoDB 写入容量。如果表已经存在，则此项不起作用	否。原定设置：`10`

此外，必须通过以下方式启用 Optimistic Concurrency Control：

hoodie.write.concurrency.mode = optimistic_concurrency_control
hoodie.cleaner.policy.failed.writes = LAZY

您可以使用 EMR 配置 API 及 hudi-defaults 分类，在集群级别无缝配置这些属性，以避免为每个作业配置此项。

删除分区

Apache Hudi 0.9.0 为其 Spark 数据源 API 引入了 DELETE_PARTITION 操作，可用于删除分区。以下是如何利用此操作的 scala 示例：

import org.apache.hudi.DataSourceWriteOptions
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.spark.sql.SaveMode

val deletePartitionDF = spark.emptyDataFrame

deletePartitionDF.write.format("hudi")
 .option(HoodieWriteConfig.TABLE_NAME, "<table name>")
 .option(DataSourceWriteOptions.OPERATION_OPT_KEY, DataSourceWriteOptions.DELETE_PARTITION_OPERATION_OPT_VAL)
 .option(DataSourceWriteOptions.PARTITIONS_TO_DELETE.key(), "<partition_value1>,<partition_value2>")
 .option(DataSourceWriteOptions.TABLE_TYPE_OPT_KEY, DataSourceWriteOptions.COW_TABLE_TYPE_OPT_VAL)
 .option(DataSourceWriteOptions.RECORDKEY_FIELD_OPT_KEY, "<record key(s)>")
 .option(DataSourceWriteOptions.PARTITIONPATH_FIELD_OPT_KEY, "<partition field(s)>") 
 .option(DataSourceWriteOptions.PRECOMBINE_FIELD_OPT_KEY, "<precombine key>")
 .mode(SaveMode.Append)
 .save("<table path>")

但是，这里有个已知问题：

由于存在错误，Hive Sync 与 DELETE_PARTITION 操作一起执行时会失败。在删除分区操作之后执行的任何以后的插入/更新插入/删除操作中，Hive Sync 都将成功。此错误已在 Hudi 0.10.0 版本中修复。

异步集群

Apache Hudi 0.9.0 通过 Spark 结构化流式传输接收器和 Delta Streamer，引入了对异步集群的支持。这样，用户就可以继续将数据提取到数据湖中，同时集群服务继续在后台运行以重新组织数据，用于提高查询性能并优化文件大小。这是通过在 Hudi 0.8.0 中引入的 Optimistic Concurrency Control 实现的。目前，只能为没有在接收任何并发更新的分区安排集群。有关如何开始使用此功能的详细信息，可以在此博文中找到。

结论

在这篇文章中，我们介绍了在 Amazon EMR 版本 5.34 和 6.5.0 及更高版本上提供的 Hudi 0.9.0 一些令人兴奋的新功能。这些新功能使得仅使用 SQL 语句即可构建数据管道，从而更轻松地在 Amazon S3 上构建事务性数据湖。

作为后续步骤，要动手体验 EMR 上的 Hudi 0.9.0，请通过此处，在使用 Amazon EMR 6.5.0 版的 EMR Studio 上试用笔记本，并告诉我们您的反馈。

亚马逊AWS官方博客