如何将PySpark中的表数据框导出到csv？

Question

我正在使用Spark 1.3.1(PySpark),我使用SQL查询生成了一个表.我现在有一个对象DataFrame.我想将此DataFrame对象(我将其称为"表")导出到csv文件,以便我可以操作它并绘制列.如何将DataFrame"表" 导出到csv文件？

谢谢!

Answer 1

如果数据框适合驱动程序内存并且您想要保存到本地文件系统,则可以使用方法将Spark DataFrame转换为本地Pandas DataFrametoPandas,然后只需使用to_csv:

df.toPandas().to_csv('mycsv.csv')

否则你可以使用spark-csv:

• Spark 1.3

  df.save('mycsv.csv', 'com.databricks.spark.csv')
  

• Spark 1.4+

  df.write.format('com.databricks.spark.csv').save('mycsv.csv')
  

在Spark 2.0+中,您可以csv直接使用数据源:

df.write.csv('mycsv.csv')

Answer 2

对于Apache Spark 2+,为了将数据帧保存到单个csv文件中.使用以下命令

query.repartition(1).write.csv("cc_out.csv", sep='|')

这1表明我只需要一个csv分区.你可以根据你的要求改变它.

Answer 3

如果您不能使用spark-csv,您可以执行以下操作:

df.rdd.map(lambda x: ",".join(map(str, x))).coalesce(1).saveAsTextFile("file.csv")

如果您需要处理带有无法使用的换行符或逗号的字符串.用这个:

import csv
import cStringIO

def row2csv(row):
    buffer = cStringIO.StringIO()
    writer = csv.writer(buffer)
    writer.writerow([str(s).encode("utf-8") for s in row])
    buffer.seek(0)
    return buffer.read().strip()

df.rdd.map(row2csv).coalesce(1).saveAsTextFile("file.csv")

Answer 4

使用 PySpark

Spark 3.0+ 中写入 csv 的最简单方法

sdf.write.csv("/path/to/csv/data.csv")

这可以根据您正在使用的 Spark 节点的数量生成多个文件。如果您想将其放在单个文件中，请使用重新分区。

sdf.repartition(1).write.csv("/path/to/csv/data.csv")

使用熊猫

如果你的数据不是太多并且可以保存在本地python中，那么你也可以使用pandas

sdf.toPandas().to_csv("/path/to/csv/data.csv", index=False)

使用考拉

sdf.to_koalas().to_csv("/path/to/csv/data.csv", index=False)

Answer 5

您需要将Dataframe重新划分为一个分区，然后以Unix文件系统格式定义文件的格式，路径和其他参数，然后就可以开始了，

df.repartition(1).write.format('com.databricks.spark.csv').save("/path/to/file/myfile.csv",header = 'true')

阅读有关重新分区功能的 更多信息阅读有关保存功能的更多信息

但是，重新分区是一项代价高昂的函数，并且toPandas（）最糟糕。尝试在以前的语法中使用.coalesce（1）代替.repartition（1）以获得更好的性能。

阅读有关分区功能与合并功能的更多信息。