如何重命名 pyspark 生成的 JSON？答案

【问题标题】：How to rename my JSON generated by pyspark?如何重命名 pyspark 生成的 JSON？
【发布时间】：2020-01-29 22:52:49
【问题描述】：

当我用

编写我的 JSON 文件时

dataframe.coalesce(1).write.format('json')

在 pyspark 上我无法更改分区中的文件名

我是这样写我的 JSON 的：

dataframe.coalesce(1).write.format('json').mode('overwrite').save('path')

但我无法更改分区中的文件名

我想要这样的路径：

/文件夹/my_name.json

'my_name.json' 是一个 json 文件

【问题讨论】：

我不认为，您可以控制输出文件的名称。您只能给出文件夹名称。

标签： json apache-spark pyspark

【解决方案1】：

在 spark we can't control name of the file 写入目录。

首先将数据写入HDFS directory，然后要更改文件名，我们需要使用HDFS api。

Example:

In Pyspark:

l=[("a",1)]
ll=["id","sa"]
df=spark.createDataFrame(l,ll)

hdfs_dir = "/folder/" #your hdfs directory
new_filename="my_name.json" #new filename

df.coalesce(1).write.format("json").mode("overwrite").save(hdfs_dir)

fs = spark._jvm.org.apache.hadoop.fs.FileSystem.get(spark._jsc.hadoopConfiguration())

#list files in the directory

list_status = fs.listStatus(spark._jvm.org.apache.hadoop.fs.Path(hdfs_dir))

#filter name of the file starts with part-

file_name = [file.getPath().getName() for file in list_status if file.getPath().getName().startswith('part-')][0]

#rename the file

fs.rename(Path(hdfs_dir+''+file_name),Path(hdfs_dir+''+new_filename))

如果要删除目录中的success files，请使用fs.delete删除_Success文件。

In Scala:

val df=Seq(("a",1)).toDF("id","sa")
df.show(false)

import org.apache.hadoop.fs._

val hdfs_dir = "/folder/"
val new_filename="new_json.json"

df.coalesce(1).write.mode("overwrite").format("json").save(hdfs_dir)

val fs=FileSystem.get(sc.hadoopConfiguration)
val f=fs.globStatus(new Path(s"${hdfs_dir}" + "*")).filter(x => x.getPath.getName.toString.startsWith("part-")).map(x => x.getPath.getName).mkString

fs.rename(new Path(s"${hdfs_dir}${f}"),new Path(s"${hdfs_dir}${new_filename}"))

fs.delete(new Path(s"${hdfs_dir}" + "_SUCCESS"))

【讨论】：

不确定如何在 python 中正确使用 java Path 类

【解决方案2】：

已接受答案的扩展。对于使用 AWS S3

的人

以下对我有用，

# Save the file to S3 bucket
spark_df.repartition(1).write.mode('append').parquet("s3://bucket_name/folder_name")

myPath = "s3://bucket_name/folder_name/*"

hadoopPath = SparkContext._jvm.org.apache.hadoop.fs.Path(myPath)

hadoopFs = hadoopPath.getFileSystem(SparkContext._jvm.org.apache.hadoop.conf.Configuration())

statuses = hadoopFs.globStatus(hadoopPath)

file_name = [file.getPath().getName() for file in statuses if file.getPath().getName().startswith('part-')][0]

hadoopFs.rename(SparkContext._jvm.org.apache.hadoop.fs.Path(f"s3://bucket_name/folder_name/{file_name}"), SparkContext._jvm.org.apache.hadoop.fs.Path("s3://bucket_name/folder_name/myFile.parquet"))

【讨论】：