如何用Glue PySpark无需加载全表删除Hudi表指定记录？

我在S3上有一张已注册到Glue Catalog的Hudi表，希望编写Glue PySpark作业删除某字段包含特定值的所有记录。
目前找到的代码示例都需要先将全表加载至DataFrame，再过滤后写回表中（示例代码如下），但我担心处理海量记录时的性能与成本问题，想知道有没有类似SQL中DELETE命令的实现方式，不用把全表加载到内存里？

原示例代码：

import sys
from awsglue.context import GlueContext
from pyspark.context import SparkContext

# Create a GlueContext
sc = SparkContext()
glueContext = GlueContext(sc)

# Get the arguments passed to the script
args = getResolvedOptions(sys.argv, ['JOB_NAME'])

# Specify the Hudi table path on Amazon S3
hudi_table_path = "s3://your-s3-bucket/hudi-table-path"

# Value to filter records on
value_to_delete = "your_value"

# Create a GlueContext and get the dynamic frame for the Hudi table
hudi_dyf = glueContext.create_dynamic_frame.from_catalog(database="your-database-name", table_name="your-table-name")

# Filter the records based on the field with the specified value
filtered_dyf = Filter.apply(frame=hudi_dyf, f=lambda x: x["your_field_name"] != value_to_delete)

# Delete the filtered records from the Hudi table
glueContext.write_dynamic_frame.from_catalog(frame=filtered_dyf, database="your-database-name", table_name="your-table-name", transformation_ctx="datasink")

# Commit the deletion operation
glueContext.commit()

# Cleanup
glueContext.cleanup()

高效实现方案 #

Hudi原生支持增量删除，无需全表加载重写，以下两种方案可以解决你的性能和成本问题：

方式1：直接使用Hudi SQL Delete语法（Glue 4.0+推荐） #

如果你的Glue作业基于Glue 4.0及以上版本（对应Spark 3.3+、Hudi 0.13+），可以直接通过Spark SQL执行DELETE语句，完全类似传统数据库的操作：

import sys
from awsglue.context import GlueContext
from pyspark.context import SparkContext
from awsglue.utils import getResolvedOptions

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session

args = getResolvedOptions(sys.argv, ['JOB_NAME'])

# 执行Hudi Delete语句，仅处理符合条件的记录
spark.sql("""
    DELETE FROM `your-database-name`.`your-table-name`
    WHERE your_field_name = 'your_value'
""")

glueContext.commit()

注意事项：

• 确保Hudi表为COPY_ON_WRITE类型（MERGE_ON_READ类型需额外配置索引同步）
• Glue 4.0+默认集成Hudi，低版本需手动引入Hudi依赖包

方式2：基于Upsert机制的增量删除（兼容旧版本） #

针对Glue 3.0及以下版本，可以通过构造带删除标记的DataFrame，仅处理需要删除的记录：

import sys
from awsglue.context import GlueContext
from pyspark.context import SparkContext
from awsglue.utils import getResolvedOptions
from pyspark.sql.functions import lit

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session

args = getResolvedOptions(sys.argv, ['JOB_NAME'])

# 1. 仅加载需要删除记录的主键（利用分区过滤或Hudi增量查询，避免全表扫描）
target_keys_df = spark.sql("""
    SELECT primary_key_column
    FROM `your-database-name`.`your-table-name`
    WHERE your_field_name = 'your_value'
""")

# 2. 为目标记录添加Hudi删除标记
delete_df = target_keys_df.withColumn("_hoodie_is_deleted", lit(True))

# 3. 执行Upsert操作，仅更新目标记录
hudi_options = {
    "hoodie.table.name": "your-table-name",
    "hoodie.datasource.write.recordkey.field": "primary_key_column",
    "hoodie.datasource.write.operation": "upsert",
    "hoodie.datasource.write.precombine.field": "primary_key_column"
}

delete_df.write.format("hudi") \
    .options(**hudi_options) \
    .mode("append") \
    .save("s3://your-s3-bucket/hudi-table-path")

glueContext.commit()

核心优势：

• 仅加载需要删除的记录主键，避免全表数据加载
• 利用Hudi的Upsert机制，仅修改目标数据文件，大幅降低IO和计算开销

原示例效率低的原因 #

你提供的示例本质是全表重写：加载全表数据→过滤掉待删除记录→重新写入全表。这种方式在数据量较大时会产生极高的存储和计算成本，完全没有利用Hudi的增量更新特性。

内容的提问来源于stack exchange，提问作者Aamit