我有一个最终的RDD labelsAndPredictions = testData.map(lambda lp: lp.label).zip(predictions)
。它的输出格式如下:
[(0.0, 0.08482142857142858), (0.0, 0.11442786069651742),.....]
我想要创建一个CSV文件,其中一列用于labels
(上面输出中的元组的第一部分),另一列用于predictions
(元组输出的第二部分)。但我不知道如何使用Python在Spark中写入CSV文件。
如何使用上述输出创建CSV文件?
发布于 2015-08-09 06:24:13
只需将RDD (labelsAndPredictions
)的行map
成字符串(CSV的行),然后使用rdd.saveAsTextFile()
。
def toCSVLine(data):
return ','.join(str(d) for d in data)
lines = labelsAndPredictions.map(toCSVLine)
lines.saveAsTextFile('hdfs://my-node:9000/tmp/labels-and-predictions.csv')
发布于 2017-02-22 06:13:05
我知道这是一个古老的帖子。但是为了帮助搜索相同内容的人,我在PySpark 1.6.2中编写了一个两列的RDD到一个CSV文件中
RDD:
>>> rdd.take(5)
[(73342, u'cells'), (62861, u'cell'), (61714, u'studies'), (61377, u'aim'), (60168, u'clinical')]
现在代码如下:
# First I convert the RDD to dataframe
from pyspark import SparkContext
df = sqlContext.createDataFrame(rdd, ['count', 'word'])
DF:
>>> df.show()
+-----+-----------+
|count| word|
+-----+-----------+
|73342| cells|
|62861| cell|
|61714| studies|
|61377| aim|
|60168| clinical|
|59275| 2|
|59221| 1|
|58274| data|
|58087|development|
|56579| cancer|
|50243| disease|
|49817| provided|
|49216| specific|
|48857| health|
|48536| study|
|47827| project|
|45573|description|
|45455| applicant|
|44739| program|
|44522| patients|
+-----+-----------+
only showing top 20 rows
现在写入CSV
# Write CSV (I have HDFS storage)
df.coalesce(1).write.format('com.databricks.spark.csv').options(header='true').save('file:///home/username/csv_out')
附言:我只是从Stackoverflow的帖子中学习的初学者。所以我不知道这是不是最好的方法。但它对我很有效,我希望它能帮助一些人!
发布于 2016-04-05 01:57:07
只用逗号连接并不好,因为如果字段包含逗号,它们将不会被正确地引起来,例如,当你想要a,b,"1,2,3",c
时,','.join(['a', 'b', '1,2,3', 'c'])
会给你a,b,1,2,3,c
。相反,您应该使用Python的csv模块将RDD中的每个列表转换为格式正确的csv字符串:
# python 3
import csv, io
def list_to_csv_str(x):
"""Given a list of strings, returns a properly-csv-formatted string."""
output = io.StringIO("")
csv.writer(output).writerow(x)
return output.getvalue().strip() # remove extra newline
# ... do stuff with your rdd ...
rdd = rdd.map(list_to_csv_str)
rdd.saveAsTextFile("output_directory")
由于csv模块只写入文件对象,因此我们必须使用io.StringIO("")
创建一个空的“文件”,并告诉csv.writer将csv格式的字符串写入其中。然后,我们使用output.getvalue()
来获取我们刚刚写入“文件”的字符串。要使此代码在Python2中工作,只需将io替换为StringIO模块。
如果您正在使用Spark,您还可以查看DataBricks save function,它具有DataFrames格式。
https://stackoverflow.com/questions/31898964
复制相似问题