时间戳

首页 > 解决方案 > 如何使用 AWS Glue 作业覆盖过时的分区数据？

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

问题描述

我有每天一次转储到s3://<bucket>/mydata/year=*/month=*/*.snappy.parquet的数据作为该月的累积数据。我有一个爬虫来爬取它以更新mydata表和一个 CW 规则，它在爬虫成功后调用 lambda，它启动 Glue 作业以转换列并输出到s3://<bucket>/mydata-transformed/year=* /月=*/*. snappy.parquet。这个流程基本上是有效的。但是，我目前遇到的问题是输出数据被附加写入而不是替换那里的数据（因为它是当月的累积数据）。例如，假设在 2020 年 10 月 1 日午夜，10/1 的数据被转储到s3://<bucket>/mydata/year=2020/month=10/*.snappy.parquet中。s3://<bucket>/mydata-transformed/year=2020/month=10/*.snappy.parquet，对于 10/1 数据来说，一切都很好。但是，第二天，当 10/1 和 10/2 的数据被转储到s3://<bucket>/mydata/year=2020/month=10/*.snappy.parquet（覆盖前一天的文件）时，Glue作业将在输出文件夹中生成附加数据，即它将包含昨天运行的数据，加上今天的运行（所以 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 年，那必须是正在处理的数据。我不太喜欢这个解决方案，但我认为它会起作用，但我不确定我还能做什么。

标签： aws-glue

解决方案

你有几个选择：

  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. 您可以使用 Gluepurge_s3_path从某个前缀中删除数据。链接在这里。

推荐阅读