将列表转换为 RDD答案

【问题标题】：Convert list to RDD将列表转换为 RDD
【发布时间】：2018-09-27 13:05:40
【问题描述】：

我正在尝试在 pyspark 中处理 samplecsv.csv 文件（64 MB）。此代码生成错误：AttributeError: 'list' object has no attribute 'saveAsTextFile'

我想我已经使用并行化将列表转换为 RDD。如果没有，它是怎么做的？

file = sc.textFile('/user/project/samplecsv.csv',5)
rdd = file.map(lambda line: (line.split(',')[0], line.split(',')[1], 
                             line.split(',')[2], line.split(',')[3], 
                             line.split(',')[4])).collect()
temp = sc.parallelize([rdd], numSlices=50000).collect()
temp.saveAsTextFile("/user/project/newfile.txt")}

【问题讨论】：

不要调用 collect() - 它会将 RDD 收集到本地列表中，这会使您的 rdd 和 temp 变量列表而不是 RDD（然后您就不需要“temp " - 无需收集然后并行化）

标签： python pyspark rdd pyspark-sql

【解决方案1】：

您的问题是您在并行列表上调用了 collect，将其返回到正常的 python 列表。

此外，您不应该在每个步骤中都调用 collect，除非您将其用于测试/调试过程。否则，您将无法利用 Spark 计算模型。

# loads the file as an rdd 
file = sc.textFile('/user/project/samplecsv.csv',5)
# builds a computation graph
rdd = file.map(lambda line: (line.split(',')[0], line.split(',')[1], 
                             line.split(',')[2], line.split(',')[3], 
                             line.split(',')[4]))
# saves the rdd to the filesystem
rdd.saveAsTextFile("/user/project/newfile.txt")

此外，您可以通过仅将行拆分一次来使代码更加优化。

【讨论】：

【解决方案2】：

我认为你应该试试下面的代码，它会解决你的目的：

file = sc.textFile("C://Users/Ravi/Desktop/test.csv",5)
rdd = file.map(lambda line: (line.split(',')[0], line.split(',')[1], 
                             line.split(',')[2], line.split(',')[3]))
rdd.coalesce(1).saveAsTextFile("C://Users/Ravi/Desktop/temp")

如果您想要分区文件，请不要使用合并。

【讨论】：