将文件写入 csv 并存储到 S3 时无法更改文件名

Question

所以我一直在研究 AWS 胶水，我在 pyspark 中创建了一个 ETL 作业，它从数据目录中读取数据并写入数据，并且 csv 存储在 S3 存储桶中。 但是每次作业运行时，它都会创建具有不同名称的不同文件，我想在每次作业运行时覆盖同一个文件。我无法找到它的正确代码。 有没有办法覆盖同一个文件（在 S3 中启用了版本控制）。

下面是代码

import sys
from awsglue.transforms
import *
from awsglue.utils
import getResolvedOptions
from pyspark.context
import SparkContext
from awsglue.context
import GlueContext
from awsglue.job
import Job


## @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

## @type: DataSource## @args: [database = "test_db", table_name = "test_dash_data", transformation_ctx = "datasource0"]## @return: datasource0## @inputs: []
datasource0 = glueContext.create_dynamic_frame.from_catalog(database = "test_db", table_name = "test_dash_data", transformation_ctx = "datasource0")

## @type: ApplyMapping## @args: [mapping = [("id", "int", "id", "int"), ("value", "int", "value", "int"), ("email", "string", "email", "string"), ("age", "int", "age", "int")], transformation_ctx = "applymapping1"]## @return: applymapping1## @inputs: [frame = datasource0]
applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("id", "int", "id", "int"), ("value", "int", "value", "int"), ("email", "string", "email", "string"), ("age", "int", "age", "int")], transformation_ctx = "applymapping1")

## @type: DataSink## @args: [connection_type = "s3", connection_options = {
  "path": "s3://auroratos3dataimport/customerdata2"
}, format = "csv", transformation_ctx = "datasink2"]
## @return: datasink2
## @inputs: [frame = applymapping1]

repartitioned1 = applymapping1.repartition(1)
datasink2 = glueContext.write_dynamic_frame.from_options(frame = repartitioned1, connection_type = "s3", connection_options = {
  "path": "s3://auroratos3dataimport/customerdata2"
}, format = "csv", transformation_ctx = "datasink2")
job.commit()

Answer 1

目前 AWS Glue 不支持 pyspark 的“覆盖”模式，但他们正在开发此功能。

不过有一个解决方法，使用普通的 pyspark：

repartitioned1.toDF()
  .write
  .mode("overwrite")
  .format("csv")
  .save(s3_path)