我有一个结果 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 文件？
【问题讨论】：
标签： python csv apache-spark pyspark file-writing
def toCSV(RDD):
for element in RDD:
return ','.join(str(element))
rows_of_csv=RDD.map(toCSV)
rows_of_csv.saveAsTextFile('/FileStore/tables/name_of_csv_file.csv')
# choose your path based on your distributed file system
【讨论】：
我知道这是一篇旧帖子。但是为了帮助搜索相同内容的人，以下是我在 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')
P.S：我只是从 Stackoverflow 中的帖子学习的初学者。所以我不知道这是否是最好的方法。但它对我有用，我希望它能帮助别人！
【讨论】：
只用逗号加入是不好的，因为如果字段包含逗号，它们将不会被正确引用，例如当你想要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() 来获取我们刚刚写入“文件”的字符串。要使此代码在 Python 2 上运行，只需将 io 替换为 StringIO 模块即可。
如果您使用的是 Spark DataFrames API，您还可以查看DataBricks save function，它具有 csv 格式。
【讨论】：
只需将map 将 RDD 的行 (labelsAndPredictions) 转换为字符串（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')
【讨论】：
csv 文件保存在哪里？在代码所在的同一目录中？我可以将其保存到其他目录（使用 saveAsTextFile('/home/files/labels-and-predictions.csv') ）吗？
part-xxx 文件。当您使用sc.textFile 时，Spark 也会读取它。对于传统工具，您可能需要先将数据合并到一个文件中。如果输出足够小，可以由传统工具处理，则没有理由通过 Spark 保存它。只需collect RDD 并在没有 Spark 的情况下将数据写入本地文件。