假设我有一个Spark DataFrame，我想将其保存为CSV文件。在Spark 2.0.0之后，DataFrameWriter类直接支持将其保存为CSV文件。
默认行为是将输出保存在提供的路径内的多个part-*.csv文件中。
如何使用以下方式保存DF：
处理此问题的一种方法是合并DF，然后保存文件。
df.coalesce(1).write.option("header", "true").csv("sample_file.csv")
然而，这种方法在收集数据时存在缺点，并且需要具备足够内存的主节点。
如果不使用 coalesce ，是否可以编写一个单一的CSV文件？如果不行，是否有比以上代码更有效的方法？
我刚刚使用pyspark和dbutils自己解决了这个问题，获取了.csv文件并将其重命名为所需的文件名。
save_location= "s3a://landing-bucket-test/export/"+year
csv_location = save_location+"temp.folder"
file_location = save_location+'export.csv'
df.repartition(1).write.csv(path=csv_location, mode="append", header="true")
file = dbutils.fs.ls(csv_location)[-1].path
dbutils.fs.cp(file, file_location)
dbutils.fs.rm(csv_location, recurse=True)
如果您只处理较小的文件并可以使用repartition(1)或coalesce(1)，则通过不使用[-1]来改善此答案，但是.csv似乎总是出现在文件夹中的最后一个。这是一种简单快速的解决方案。
使用：
df.toPandas().to_csv("sample_file.csv", header=True)
df.coalesce(1).write.option("inferSchema","true").csv("/newFolder",header =
'true',dateFormat = "yyyy-MM-dd HH:mm:ss")
以下的scala方法适用于本地或客户端模式，并将df写入所选择的单个csv文件。它要求df适合内存，否则collect()会崩溃。
import org.apache.hadoop.fs.{FileSystem, Path}
val SPARK_WRITE_LOCATION = some_directory
val SPARKSESSION = org.apache.spark.sql.SparkSession
def saveResults(results : DataFrame, filename: String) {
var fs = FileSystem.get(this.SPARKSESSION.sparkContext.hadoopConfiguration)
if (SPARKSESSION.conf.get("spark.master").toString.contains("local")) {
fs = FileSystem.getLocal(new conf.Configuration())
}
val tempWritePath = new Path(SPARK_WRITE_LOCATION)
if (fs.exists(tempWritePath)) {
val x = fs.delete(new Path(SPARK_WRITE_LOCATION), true)
assert(x)
}
if (results.count > 0) {
val hadoopFilepath = new Path(SPARK_WRITE_LOCATION, filename)
val writeStream = fs.create(hadoopFilepath, true)
val bw = new BufferedWriter( new OutputStreamWriter( writeStream, "UTF-8" ) )
val x = results.collect()
for (row : Row <- x) {
val rowString = row.mkString(start = "", sep = ",", end="\n")
bw.write(rowString)
}
bw.close()
writeStream.close()
val resultsWritePath = new Path(WRITE_DIRECTORY, filename)
if (fs.exists(resultsWritePath)) {
fs.delete(resultsWritePath, true)
}
fs.copyToLocalFile(false, hadoopFilepath, resultsWritePath, true)
} else {
System.exit(-1)
}
}这个解决方案基于Shell脚本，没有并行化，但在SSD上仍然非常快。它使用Unix系统上的cat和输出重定向。假设包含分区的CSV目录位于/my/csv/dir，输出文件为/my/csv/output.csv：
#!/bin/bash
echo "col1,col2,col3" > /my/csv/output.csv
for i in /my/csv/dir/*.csv ; do
echo "Processing $i"
cat $i >> /my/csv/output.csv
rm $i
done
echo "Done"
在将每个分区附加到最终CSV后，它将删除每个分区以释放空间。
"col1,col2,col3"是CSV文件的标题（这里有三列名称为col1、col2和col3）。您必须告诉Spark不要在每个分区中放置标题（可以使用.option("header", "false")来实现此目的，因为Shell脚本会处理它）。
如果您仍然想要这样做，以下是我在使用Spark 2.1和一些java.nio.file帮助的Scala中完成此操作的方法。
基于https://fullstackml.com/how-to-export-data-frame-from-apache-spark-3215274ee9d6
val df: org.apache.spark.sql.DataFrame = ??? // data frame to write
val file: java.nio.file.Path = ??? // target output file (i.e. 'out.csv')
import scala.collection.JavaConversions._
// write csv into temp directory which contains the additional spark output files
// could use Files.createTempDirectory instead
val tempDir = file.getParent.resolve(file.getFileName + "_tmp")
df.coalesce(1)
.write.format("com.databricks.spark.csv")
.option("header", "true")
.save(tempDir.toAbsolutePath.toString)
// find the actual csv file
val tmpCsvFile = Files.walk(tempDir, 1).iterator().toSeq.find { p =>
val fname = p.getFileName.toString
fname.startsWith("part-00000") && fname.endsWith(".csv") && Files.isRegularFile(p)
}.get
// move to desired final path
Files.move(tmpCsvFile, file)
// delete temp directory
Files.walk(tempDir)
.sorted(java.util.Comparator.reverseOrder())
.iterator().toSeq
.foreach(Files.delete(_))
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs._
def merge(srcPath: String, dstPath: String): Unit = {
val hadoopConfig = new Configuration()
val hdfs = FileSystem.get(hadoopConfig)
FileUtil.copyMerge(hdfs, new Path(srcPath), hdfs, new Path(dstPath), true, hadoopConfig, null)
// the "true" setting deletes the source files once they are merged into the new output
}
请查看使用spark-csv编写单个CSV文件的方法。
coalesce(1)或repartition(1)之外的spark方法。如果您想让多个工作人员附加到同一个文件，则它们必须顺序执行或等待彼此完成，否则记录将无序，这将难以编排。 - Davospart-*的单个文件。 - Spandan Brahmbhatt