TypeError: Objekt vom Typ 'PipelinedRDD' hat keine len()

Question

alle! Ich treffe diese Protokollfehler, wenn ein Python-Code in Funken

Code ausgeführt wird:

Haupt 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 

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

wenn Funken -submit Haupt Py Ausgangsfehlerprotokoll:

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

wie es zu lösen !! danke !!

Answer 1

newWordCountDictList ist RDD (verteiltes Objekt und in mehreren Arbeitsknoten) Objekt nicht lokales Objekt Sammlung in Ihrem Treiberprogramm.

können Sie entweder

n = newWordCountDictList.count() 

oder

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

korrektes Ergebnis zu erhalten.