我已经用databrick csv包启动了shell
#../spark-1.6.1-bin-hadoop2.6/bin/pyspark --packages com.databricks:spark-csv_2.11:1.3.0
然后我读了一个csv文件,进行了一些分组操作,并将其转储到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有没有一种方法可以创建一个csv文件(不再是目录),可以通过msoffice或记事本++打开???
注意:我目前没有使用集群,因为它对像我这样的spark初学者来说太复杂了。如果有人能提供一个链接,说明如何在集群环境中处理tocsv到单个文件中,那将是一个很大的帮助
尝试
df.coalesce(1).write.format('com.databricks.spark.csv').save('path+my.csv',header = 'true')
请注意,这可能不是您当前设置的问题,但在超大数据集上,您可能会在驱动程序上遇到内存问题。这也需要更长的时间(在集群场景中),因为所有东西都必须推回到单个位置。
以防万一,在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")
使用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
.write
.mode('overwrite')
.option('header', 'true')
.csv('output.csv')
或者,如果希望结果在单个分区中,可以使用coalesce(1)
:
df
.coalesce(1)
.write
.mode('overwrite')
.option('header', 'true')
.csv('output.csv')
然而,请注意,这是一项昂贵的操作,对于超大的数据集可能不可行。
得到了第一个问题的答案,这是一个将一个额外的参数header='true'与csv语句一起传递的问题
df.write.format('com.databricks.spark.csv').save('path+my.csv',header = 'true')
#第2个问题的备选方案
使用topandas.to_csv,但我再次不想在这里使用熊猫,所以请建议是否有其他方法。