AWS Glue 输出文件名

Question

我正在使用 AWS 来转换一些 JSON 文件。我已将文件从 S3 添加到 Glue。我设置的作业读取文件正常，作业运行成功，有一个文件添加到正确的 S3 存储桶中。我遇到的问题是我无法命名文件 - 它被赋予了一个随机名称，它也没有被赋予 .JSON 扩展名。

如何命名文件并将扩展名添加到输出中？

Answer 1

由于 Spark 工作方式的性质，无法命名文件。但是，之后可以立即重命名文件。

URI = sc._gateway.jvm.java.net.URI
Path = sc._gateway.jvm.org.apache.hadoop.fs.Path
FileSystem = sc._gateway.jvm.org.apache.hadoop.fs.FileSystem
fs = FileSystem.get(URI("s3://{bucket_name}"), sc._jsc.hadoopConfiguration())

file_path = "s3://{bucket_name}/processed/source={source_name}/year={partition_year}/week={partition_week}/"
df.coalesce(1).write.format("json").mode(
    "overwrite").option("codec", "gzip").save(file_path)

# rename created file
created_file_path = fs.globStatus(Path(file_path + "part*.gz"))[0].getPath()
fs.rename(
    created_file_path,
    Path(file_path + "{desired_name}.jl.gz"))

Answer 2

以下代码对我有用 -

source_DataFrame = glueContext.create_dynamic_frame.from_catalog(database = databasename, table_name = source_tablename_in_catalog, transformation_ctx = "source_DataFrame")

source_DataFrame = source_DataFrame.toDF().coalesce(1) #avoiding coalesce(1) will create many part-000* files according to data

from awsglue.dynamicframe import DynamicFrame
DyF = DynamicFrame.fromDF(source_DataFrame, glueContext, "DyF")

# writing the file as usual in Glue. **I have given some partitions** too.
# keep "partitionKeys":[] in case of no partitions
output_Parquet = glueContext.write_dynamic_frame.from_options(frame = DyF, connection_type = "s3", format = "parquet", connection_options = {"path": destination_path + "/", "partitionKeys": ["department","team","card","datepartition"]}, transformation_ctx = "output_Parquet")

import boto3
client = boto3.client('s3')

#getting all the content/file inside the bucket. 
response = client.list_objects_v2(Bucket=bucket_name)
names = response["Contents"]

#Find out the file which have part-000* in it's Key
particulars = [name['Key'] for name in names if 'part-000' in name['Key']]

#Find out the prefix of part-000* because we want to retain the partitions schema 
location = [particular.split('part-000')[0] for particular in particulars]

#Constrain - copy_object has limit of 5GB.datepartition=20190131
for key,particular in enumerate(particulars):
    client.copy_object(Bucket=bucket_name, CopySource=bucket_name + "/" + particular, Key=location[key]+"newfile")
    client.delete_object(Bucket=bucket_name, Key=particular)

job.commit()

基石是当文件（copy_object）大于 5GB 时，它会复制失败。 你可以用这个

s3 = boto3.resource('s3')
for key,particular in enumerate(particulars):
    copy_source = {
        'Bucket': bucket_name,
        'Key': particular
    }
    s3.meta.client.copy(copy_source, bucket_name, location[key]+"newfile")