如何在磁盘上将火花DataFrame保存为csv？

Question

例如,结果如下:

df.filter("project = 'en'").select("title","count").groupBy("title").sum()

会返回一个数组.

如何将火花DataFrame保存为磁盘上的csv文件？

Answer 1

Apache Spark不支持磁盘上的本机CSV输出.

您有四种可用的解决方案:

1. 您可以将Dataframe转换为RDD:

   def convertToReadableString(r : Row) = ???
   df.rdd.map{ convertToReadableString }.saveAsTextFile(filepath)
   

   这将创建一个文件夹文件路径.在文件路径下,您将找到分区文件(例如part-000*)

   如果我想将所有分区附加到一个大的CSV中,我通常会这样做

   cat filePath/part* > mycsvfile.csv
   

   一些将用于coalesce(1,false)从RDD创建一个分区.这通常是一种不好的做法,因为它可能会通过将您收集的所有数据提取到驱动程序来压倒驱动程序.

   请注意,df.rdd将返回一个RDD[Row].

2. 使用Spark <2,您可以使用databricks spark-csv 库:

   • Spark 1.4+:

     df.write.format("com.databricks.spark.csv").save(filepath)
     

   • Spark 1.3:

     df.save(filepath,"com.databricks.spark.csv")
     

3. 使用Spark 2.x时,spark-csv不需要包,因为它包含在Spark中.

   df.write.format("csv").save(filepath)
   

4. 您可以转换为本地Pandas数据框和使用to_csv方法(仅限PySpark).

注意:解决方案1,2和3将导致CSV格式文件(part-*)由Spark调用时调用的底层Hadoop API生成save.part-每个分区只有一个文件.

Answer 2

将数据帧作为 csv 写入磁盘与从 csv 读取类似。如果您希望将结果作为一个文件，您可以使用 coalesce。

df.coalesce(1)
      .write
      .option("header","true")
      .option("sep",",")
      .mode("overwrite")
      .csv("output/path")

如果你的结果是一个数组，你应该使用特定于语言的解决方案，而不是 spark dataframe api。因为所有这些结果都返回驱动程序机器。