如何使用 AWS Glue 作业覆盖过时的分区数据？

Question

我的数据每天转储一次到 s3:///mydata/year=*/month=*/*.snappy.parquet 作为该月的累积数据。我有一个爬虫来爬取它以更新 mydata 表和一个 CW 规则，该规则在爬虫成功后调用 lambda，它启动 Glue 作业以转换列并输出到 s3:///mydata-transformed/year=*/month=*/*.snappy.parquet。这个流程基本上是有效的。但是，我目前遇到的问题是输出数据被附加写入而不是替换那里的数据（因为它是当月的累积数据）。例如，假设在 2020 年 10 月 1 日午夜，10/1 的数据被转储到 s3:///mydata/year=2020/month=10/*.snappy.parquet .该流程将在 s3:///mydata-transformed/year=2020/month=10/*.snappy.parquet 中生成转换后的数据，一切都适用于 10/1 数据。但是，第二天 10/1 和 10/2 的数据被转储到 s3:///mydata/year=2020/month=10/*.snappy.parquet（覆盖前一天的文件），胶水作业将在输出文件夹中生成附加数据，即它将包含昨天运行的数据，加上今天的运行（所以 10/1 数据两次，10/2 数据）。第二天，10/1 数据 3X、10/2 数据 2X 和 10/3 数据。等等。 2020/09 年及之前的数据没问题，因为它们没有变化。下面是我的代码的基本结构，去掉了样板代码，用人为的替换了真正的转换。

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)
DataSource0 = glueContext.create_dynamic_frame.from_catalog(database = "mydatabase", table_name = "mydata", transformation_ctx = "DataSource0")
ds_df = DataSource0.toDF()
ds_df1 = ds_df.select("year","month",upper(col('colA')),upper(col('colB')),upper(col('colC')),upper(col('colD')))
Transform0 = DynamicFrame.fromDF(ds_df1, glueContext, "Transform0")
DataSink0 = glueContext.write_dynamic_frame.from_options(frame = Transform0, connection_type = "s3", format = "parquet", connection_options = {"path": "s3://<bucket>/mydata-transformed/", "partitionKeys": ["year","month"]}, transformation_ctx = "DataSink0")
job.commit()

我该怎么做才能使当月前一天的数据被删除并替换为当前工作的数据？有没有办法知道，在我的示例中，源数据中的 month=10 分区已更改，因此我可以在进行转换之前清除输出中的相同分区并输出新的数据？

谢谢。

[编辑] 因此，似乎一种解决方案是获取作业书签，然后使用 CURR_LATEST_PARTITIONS 值来确定在处理数据之前我应该​​删除哪个分区。就我而言，当我处理 2020/10 时，CURR_LATEST_PARTITIONS 是 2020/09。所以我知道要删除 2020/10 年的数据，因为如果 CURR_LATEST_PARTITIONS 是 2020/09 年，那必须是正在处理的数据。我不太喜欢这个解决方案，但我认为它会起作用，但我不确定我还能做什么。

Answer 1

你有几个选择：

1. DynamicFrameWriter 尚不支持覆盖 S3 中的数据。相反，您可以使用 Spark 原生 write()。但是，对于非常大的数据集，它可能会有点低效，因为单个工作人员将用于覆盖 S3 中的现有数据。下面是一个例子：

    sc = SparkContext()
    glueContext = GlueContext(sc)
    spark = glueContext.spark_session
    spark.conf.set("spark.sql.sources.partitionOverwriteMode","dynamic")
    
    job = Job(glueContext)
    job.init(args['JOB_NAME'], args)
    
    DataSource0 = glueContext.create_dynamic_frame.from_catalog(database = "mydatabase", table_name = "mydata", transformation_ctx = "DataSource0")
    ds_df = DataSource0.toDF()
    ds_df1 = ds_df.select("year","month",upper(col('colA')),upper(col('colB')),upper(col('colC')),upper(col('colD')))
    ds_df1 \
        .write.mode('overwrite') \
        .format('parquet') \
        .partitionBy('year', 'month') \
        .save('s3://<bucket>/mydata-transformed/')
    
    job.commit()

1. 在 lambda 函数中，您可以在 S3 中使用特定前缀下的删除数据。使用 Python 和 boto3 的示例是：

    import boto3
    
    s3_res = boto3.resource('s3')
    bucket = 'my-bucket-name'
    # Add any logic to derive required prefix based on year/month/day
    prefix = 'mydata/year=2020/month=10/'
    s3_res.Bucket(bucket).objects.filter(Prefix=key).delete()

1. 您可以使用Glue的purge_s3_path删除某个前缀的数据。链接是here。

Answer 2

现在胶水中存在删除 S3 路径或删除胶水目录表的功能。

AWS GLue doc