通过胶水作业将拼花地板写入s3时作业中止

我的代码如下所示,其中包括转换:

dictionaryDf = spark.read.option("header", "true").csv(
            "s3://...../.csv")

        web_notif_data = fullLoad.cache()
        web_notif_data.persist(StorageLevel.MEMORY_AND_DISK)
        print("::::::data has been loaded::::::::::::")
        distinct_campaign_name = web_notif_data.select(
            trim(web_notif_data.campaign_name).alias("campaign_name")).distinct()
        web_notif_data.createOrReplaceTempView("temp")
        variablesList = Config.get('web', 'variablesListWeb')
        web_notif_data = spark.sql(variablesList)
        web_notif_data.persist(StorageLevel.MEMORY_AND_DISK)
        web_notif_data = web_notif_data.withColumn("camp", regexp_replace("campaign_name", "_", ""))
        web_notif_data = web_notif_data.drop("campaign_name")
        web_notif_data = web_notif_data.withColumnRenamed("camp", "campaign_name")
        web_notif_data = web_notif_data.withColumn("channel", lit("web_notification"))
        web_notif_data.createOrReplaceTempView("data")
        campaignTeamWeb = Config.get('web', 'campaignTeamWeb')
        web_notif_data = spark.sql(campaignTeamWeb)
        web_notif_data.persist(StorageLevel.MEMORY_AND_DISK)

        distinct_campaign_name = distinct_campaign_name.withColumn("camp", F.regexp_replace(
            F.lower(F.trim(col("campaign_name"))),
            "[^a-zA-Z0-9]", ""))
        output_df3 = (
            distinct_campaign_name.withColumn("cname_split",
                                              F.explode(F.split(F.lower(F.trim(col("campaign_name"))), "_")))
                .join(
                dictionaryDf,
                (
                        (
                                (F.col("function") == "contains") &
                                F.col("camp").contains(F.col("terms"))
                        ) |
                        (
                                (F.col("function") == "match") &
                                F.col("campaign_name").contains("_") &
                                (F.col("cname_split") == F.col("terms"))
                        )
                ),
                "left"
            )
                .withColumn(
                "empty_is_other",
                F.when(
                    (
                            F.col("product").isNull() &
                            F.col("product_category").isNull()
                    ),
                    "other"
                )
            )
                .withColumn(
                "rn",
                F.row_number().over(
                    Window.partitionBy("campaign_name")
                        .orderBy(
                        F.when(
                            F.col("function").isNull(), 3
                        ).when(
                            F.col("function") == "match", 2
                        ).otherwise(1),
                        F.length(F.col("terms")).desc(),
                        F.col("product").isNull()
                    )
                )
            )
                .filter("rn=1")
                .select(
                "campaign_name",
                F.coalesce("product", "empty_is_other").alias("prod"),
                F.coalesce("product_category", "empty_is_other").alias("prod_cat"),
            )
                .na.fill("")
        )
        print(":::::::::::transformations have been done finally::::::::::::")
        web_notif_data1 = web_notif_data  # Just taking the backup of DF in case something goes wrong
        web_notif_data = web_notif_data.drop("campaign_name")
        web_notif_data = web_notif_data.withColumnRenamed("temp_campaign_name", "campaign_name")
        veryFinalDF = web_notif_data.join(output_df3, "campaign_name", "left_outer")
        # veryFinalDF.show(truncate=False)
        veryFinalDF.write.mode("overwrite").parquet(aggregatedPath)
        print("::::final data have been written successfully::::::")

哪里 fullLoad 是从红移表读取的数据帧。这段代码在计算机上运行良好 0.2 Million 记录。然而,在15天的生产过程中,数据可能在一周左右 minimum 属于 25 Million 记录。我不知道大小,因为数据存储在红移表中,我们从中读取数据,然后处理数据。我通过 Glue 它被卡在最后一行,也就是说,当把数据写成拼花地板的时候。它给出了以下错误:

我试着用 30 遗嘱执行人。它需要大约 20 mins 从中加载数据 Redshift 进入 满载 数据帧。还有什么可以避免这个错误呢?我是AWS和胶水行业的新手。