将列表转换为RDD

Question

我正在努力示例csv.csvpyspark中的文件（64 MB）。 此代码生成错误：AttributeError:'list'对象没有属性“saveAsTextFile”

我想我已经用parallelize把list转换成了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")}

Answer 1

我认为您应该尝试以下代码，它将解决您的目的：

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")

如果需要分区文件，请不要使用coalesce。在

Answer 2

您的问题是对并行化列表调用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")

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