从 Apache Hudi 表中删除记录，该表是使用 AWS Glue Job 和 Kinesis 创建的 Glue 表的一部分

Question

我目前配置了一个 DynamoDB 流，它在插入/更新发生时将流输入到 Kinesis 数据流中，随后我有 Glue 表，它从上面的运动流中获取输入，然后显示结构模式，还有一个 Glue 脚本帮助我创建一个可以使用 Athena 访问的 Hudi 表。我目前能够监控流数据并能够在我的 Athena 表中看到插入/更新（在我的本地机器上使用 pycharm 从 boto3 模拟）。我们可以使用相同的 Glue Job 执行删除吗？

我的胶水作业如下所示 -

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.sql.session import SparkSession
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.sql import DataFrame, Row
from pyspark.sql.functions import * 
from pyspark.sql.functions import col, to_timestamp, monotonically_increasing_id, to_date, when
import datetime
from awsglue import DynamicFrame

import boto3

## @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ["JOB_NAME", "database_name", "kinesis_table_name", "starting_position_of_kinesis_iterator", "hudi_table_name", "window_size", "s3_path_hudi", "s3_path_spark" ])

spark = SparkSession.builder.config('spark.serializer','org.apache.spark.serializer.KryoSerializer').config('spark.sql.hive.convertMetastoreParquet','false').getOrCreate()
                    
sc = spark.sparkContext
glueContext = GlueContext(sc)
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

database_name = args["database_name"]
kinesis_table_name = args["kinesis_table_name"]
hudi_table_name = args["hudi_table_name"]
s3_path_hudi = args["s3_path_hudi"]
s3_path_spark = args["s3_path_spark"]

commonConfig = {'hoodie.datasource.write.hive_style_partitioning' : 'true','className' : 'org.apache.hudi', 'hoodie.datasource.hive_sync.use_jdbc':'false', 'hoodie.datasource.write.precombine.field': 'id', 'hoodie.datasource.write.recordkey.field': 'id', 'hoodie.table.name': hudi_table_name, 'hoodie.consistency.check.enabled': 'true', 'hoodie.datasource.hive_sync.database': database_name, 'hoodie.datasource.hive_sync.table': hudi_table_name, 'hoodie.datasource.hive_sync.enable': 'true', 'path': s3_path_hudi}

partitionDataConfig = { 'hoodie.datasource.write.keygenerator.class' : 'org.apache.hudi.keygen.ComplexKeyGenerator', 'hoodie.datasource.write.partitionpath.field': "partitionkey, partitionkey2 ", 'hoodie.datasource.hive_sync.partition_extractor_class': 'org.apache.hudi.hive.MultiPartKeysValueExtractor', 'hoodie.datasource.hive_sync.partition_fields': "partitionkey, partitionkey2"}

incrementalConfig = {'hoodie.upsert.shuffle.parallelism': 68, 'hoodie.datasource.write.operation': 'upsert', 'hoodie.cleaner.policy': 'KEEP_LATEST_COMMITS', 'hoodie.cleaner.commits.retained': 2}

combinedConf = {**commonConfig, **partitionDataConfig, **incrementalConfig}

glue_temp_storage = s3_path_hudi

data_frame_DataSource0 = glueContext.create_data_frame.from_catalog(database = database_name, table_name = kinesis_table_name, transformation_ctx = "DataSource0", additional_options = {"startingPosition": "TRIM_HORIZON", "inferSchema": "true"})

def processBatch(data_frame, batchId):
    if (data_frame.count() > 0):

        DataSource0 = DynamicFrame.fromDF(data_frame, glueContext, "from_data_frame")
        
        your_map = [
            ('eventName', 'string', 'eventName', 'string'),
            ('userIdentity', 'string', 'userIdentity', 'string'),
            ('eventSource', 'string', 'eventSource', 'string'),
            ('tableName', 'string', 'tableName', 'string'),
            ('recordFormat', 'string', 'recordFormat', 'string'),
            ('eventID', 'string', 'eventID', 'string'),
            ('dynamodb.ApproximateCreationDateTime', 'long', 'ApproximateCreationDateTime', 'long'),
            ('dynamodb.SizeBytes', 'long', 'SizeBytes', 'long'),
            ('dynamodb.NewImage.id.S', 'string', 'id', 'string'),
            ('dynamodb.NewImage.custName.S', 'string', 'custName', 'string'),
            ('dynamodb.NewImage.email.S', 'string', 'email', 'string'),
            ('dynamodb.NewImage.registrationDate.S', 'string', 'registrationDate', 'string'),
            ('awsRegion', 'string', 'awsRegion', 'string')
        ]

        new_df = ApplyMapping.apply(frame = DataSource0, mappings=your_map, transformation_ctx = "applymapping1")
        abc = new_df.toDF()
        
        inputDf = abc.withColumn('update_ts_dms',to_timestamp(abc["registrationDate"])).withColumn('partitionkey',abc["id"].substr(-1,1)).withColumn('partitionkey2',abc["id"].substr(-2,1))
        
        

        # glueContext.write_dynamic_frame.from_options(frame = DynamicFrame.fromDF(inputDf, glueContext, "inputDf"), connection_type = "marketplace.spark", connection_options = combinedConf)
        glueContext.write_dynamic_frame.from_options(frame = DynamicFrame.fromDF(inputDf, glueContext, "inputDf"), connection_type = "custom.spark", connection_options = combinedConf)


glueContext.forEachBatch(frame = data_frame_DataSource0, batch_function = processBatch, options = {"windowSize": "10 seconds", "checkpointLocation":  s3_path_spark})


job.commit()

如何实施删除/脚本以在同一脚本中反映我的 dynamodb 表中的删除更改？有可能吗？

编辑：

在脚本中添加了以下行。导致删除 Athena 中的所有行和空表。

deleteDataConfig = {'hoodie.datasource.write.operation': 'delete'}


combinedConf = {**commonConfig, **partitionDataConfig, **incrementalConfig, **deleteDataConfig}

Answer 1

Hudi版本大于0.5.1可以实现删除

df.write.format("org.apache.hudi").
options(getQuickstartWriteConfigs).
option(OPERATION_OPT_KEY,"delete").
option(PRECOMBINE_FIELD_OPT_KEY, "ts").
option(RECORDKEY_FIELD_OPT_KEY, "uuid").
option(PARTITIONPATH_FIELD_OPT_KEY, "partitionpath").
option(TABLE_NAME, tableName).
mode(Append).
save(basePath);

Reference Reference