Ошибка - pyspark, пересекающий вложенный список и список RDD

Question

У меня есть список списков в RDD и список для пересечения.B необходимо взаимодействовать с каждым списком в A.

A = [[a,b,c,d],[e,f,g,h]....]
B = [a,b,c,d,e,f,g,h]

Мне нужно пересечь эти два, чтобы получить общие буквы.Я использовал следующее, но получил ошибку из-за typeError

pwords = A.intersection(B)

Затем я попытался использовать распараллеливание, основываясь на нескольких предложениях по stackoverflow, но получил ошибку.

text_words = sc.parallelize(A)
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/usr/local/spark/python/pyspark/context.py", line 501, in 
parallelize
c = list(c)    # Make it a list so we can compute its length
TypeError: 'PipelinedRDD' object is not iterable

Когда я попытался преобразоватьв список, как показано в сообщении об ошибке.Я снова получил ошибку.

TypeError: 'PipelinedRDD' object is not iterable 

Я пытался следовать Найти пересечение двух вложенных списков? и получил эту ошибку:

TypeError: 'PipelinedRDD' object is not iterable

Answer 1

Я не уверен на 100% в вашем вопросе, но, думаю, у вас есть вложенный список в качестве СДР и вы хотите пересечь его со статическим списком B. Затем каждый элемент во вложенных списках должен быть проверен на наличие в B иесли он существует, он должен остаться.

Если порядок элементов не имеет значения, вы можете использовать следующий код:

A = [["a","b","c","d"],["e","f","g","h"],["i","j","k","l"],["a","b","x","f","y"]]
B = ["a","b","c","d","e","f","g","h"]

text_words = sc.parallelize(A)
text_words.map(lambda x: list(set(x) & set(B))).collect()

Выход:

[['a', 'c', 'b', 'd'], ['h', 'e', 'g', 'f'], [], ['a', 'b', 'f']]