CODEWITHSUNDEEP
pyspark
Peihua Zhang
Peihua Zhang

Reputation: 29

TypeError: object of type 'PipelinedRDD' has no len()

everyone! I meet this error log when running a python code in spark

code :

main.py

sc = SparkContext(appName="newsCluster")
sc.addPyFile("/home/warrior/gitDir/pysparkCode/clusterNews/wordSeg.py")
sc.addPyFile("/home/warrior/gitDir/pysparkCode/clusterNews/sparkClusterAlgorithm.py")
wordseg = wordSeg()
clustermanage = sparkClusterAlgorithm()
files = sc.wholeTextFiles("hdfs://warrior:9000/testData/skynews")
file_list = files.map(lambda item: item[1])
file_wc_dict_list = file_list.map(lambda file_content:wordseg.GetWordCountTable(file_content))
file_wc_dict_list.persist()
all_word_dict = wordseg.updateAllWordList(file_wc_dict_list)

wordSeg.py

def updateAllWordList(self, newWordCountDictList):
    '''
        description: input an new file then update the all word list
        input:
            newWordCountDict: new input string word count dict
        output:
            all_word_dict
    '''
    n = len(newWordCountDictList)
    all_word_list = []
    all_word_dict = {}
    for i in range(0,n):
        all_word_list = list(set(all_word_list + newWordCountDictList[i].keys()))
    for i in range(0,len(all_word_list)):
        all_word_dict[all_word_list[i]]=0
    return all_word_dict

....... .......

when spark-submit main.py output error log:

Traceback (most recent call last):                                              
File "/home/warrior/gitDir/pysparkCode/clusterNews/__main__.py", line 31, in <module>
  all_word_dict =     wordseg.updateAllWordList(file_wc_dict_list)#file_wc_dict_list.map(lambda     file_wc_dict:wordseg.updateAllWordList(file_wc_dict))
File "/home/warrior/gitDir/pysparkCode/clusterNews/wordSeg.py", line 54, in updateAllWordList
  n = len(newWordCountDictList)
TypeError: object of type 'PipelinedRDD' has no len()

how to solve it!! thanks!!

Upvotes: 2

Views: 6780

Answers (1)

Shawn Guo
Shawn Guo

Reputation: 3228

newWordCountDictList is RDD(distributed object and located in multiple work nodes) object not local collection object in your driver program.

You can use either

n = newWordCountDictList.count()

or 

all_word_dict = wordseg.updateAllWordList(file_wc_dict_list.collect())

to get correct result.

Upvotes: 4

Related Questions