使用列名编写 csv 并读取从 Pyspark 中的 sparksql 数据帧生成的 csv 文件答案

【问题标题】：writing a csv with column names and reading a csv file which is being generated from a sparksql dataframe in Pyspark使用列名编写 csv 并读取从 Pyspark 中的 sparksql 数据帧生成的 csv 文件
【发布时间】：2016-12-01 08:57:20
【问题描述】：

我已经用 databrick csv 包启动了 shell

#../spark-1.6.1-bin-hadoop2.6/bin/pyspark --packages com.databricks:spark-csv_2.11:1.3.0

然后我读取了一个 csv 文件，做了一些 groupby 操作并将其转储到 csv。

from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)
df = sqlContext.read.format('com.databricks.spark.csv').options(header='true').load(path.csv')   ####it has columns and df.columns works fine
type(df)   #<class 'pyspark.sql.dataframe.DataFrame'>
#now trying to dump a csv
df.write.format('com.databricks.spark.csv').save('path+my.csv')
#it creates a directory my.csv with 2 partitions
### To create single file i followed below line of code
#df.rdd.map(lambda x: ",".join(map(str, x))).coalesce(1).saveAsTextFile("path+file_satya.csv") ## this creates one partition in directory of csv name
#but in both cases no columns information(How to add column names to that csv file???)
# again i am trying to read that csv by
df_new = sqlContext.read.format("com.databricks.spark.csv").option("header", "true").load("the file i just created.csv")
#i am not getting any columns in that..1st row becomes column names

请不要在 read_csv 之后或在阅读提及列名时向数据框添加架构。

问题 1- 在提供 csv 转储时，有什么方法可以添加列名？？？

问题2-有没有办法创建可以由ms office或notepad++打开的单个csv文件（不是目录）？？？

注意：我目前没有使用集群，因为它对于像我这样的 spark 初学者来说太复杂了。如果有人能提供如何在集群环境中将 to_csv 处理成单个文件的链接，那将是一个很大的帮助。

【问题讨论】：

标签： python apache-spark pyspark apache-spark-sql pyspark-sql

【解决方案1】：

得到第一个问题的答案，这是一个额外的参数 header = 'true' 以及 csv 语句的问题

df.write.format('com.databricks.spark.csv').save('path+my.csv',header = 'true')

#第二个问题的替代方案

使用 topandas.to_csv ，但我不想在这里使用 pandas，所以请建议是否有其他方法。

【讨论】：

【解决方案2】：

试试

df.coalesce(1).write.format('com.databricks.spark.csv').save('path+my.csv',header = 'true')

请注意，这在您当前的设置中可能不是问题，但在非常大的数据集上，您可能会遇到驱动程序的内存问题。这也将花费更长的时间（在集群场景中），因为一切都必须推回一个位置。

【讨论】：

@Mike-是的，我正在使用一个巨大的数据集，但是数据集（在对那个大数据集执行一些功能之后），我想要 o/p，因为 csv 可能有 100 万行或更少.我有 28GB RAM（在主服务器和它的两个从服务器上）。我一定会尝试检查它是否给我内存错误。只是为了好奇，如果我想 o/p 一个大约 500 万行的 csv，你能建议什么是理想的配置吗？
@Satya - 如果我需要这种格式，我主要使用 Spark 之外的其他工具（即 cat、gzip 等）重新组合文件。关于最佳配置，这取决于您尝试使用什么来读取文件。我的大部分使用是预处理，然后重新导入 SQL 数据库以进行实时查询 - 运行批量导入不需要单个文件。

【解决方案3】：

使用 spark >= 2.o，我们可以做类似的事情

df = spark.read.csv('path+filename.csv', sep = 'ifany',header='true')
df.write.csv('path_filename of csv',header=True) ###yes still in partitions
df.toPandas().to_csv('path_filename of csv',index=False)  ###single csv(Pandas Style)

【讨论】：

需要注意的是，您可以通过执行df.coalesce(1).write.csv(..., header = True) 来强制使用单个csv。如果您对 csv 进行分区，这将为每个分区创建一个文件。输出文件的名称将是 gobbledygook。

【解决方案4】：

以防万一，在 spark 2.1 上，您可以使用以下行创建单个 csv 文件

dataframe.coalesce(1) //So just a single part- file will be created
.write.mode(SaveMode.Overwrite)
.option("mapreduce.fileoutputcommitter.marksuccessfuljobs","false") //Avoid creating of crc files
.option("header","true") //Write the header
.csv("csvFullPath")

【讨论】：

【解决方案5】：

以下应该可以解决问题：

df \
  .write \
  .mode('overwrite') \
  .option('header', 'true') \
  .csv('output.csv')

或者，如果您希望结果在单个分区中，您可以使用coalesce(1)：

df \
  .coalesce(1) \
  .write \
  .mode('overwrite') \
  .option('header', 'true') \
  .csv('output.csv')

但是请注意，这是一项昂贵的操作，并且对于非常大的数据集可能不可行。

【讨论】：