Как разбить пары ключ / последовательность в СДР Spark на пары ключ / значение?

Question

У меня есть СДР в этом формате:

(key, [v1, v2, v3, ..., vn])

, и я хочу преобразовать его в СДР следующим образом:

(key, v1), (key, v2), ..., (key, vn)

Для уточнения, я использую sc.wholeTextFile(dir) для чтения всех текстовых документов в каталоге dir, и СДР будет выглядеть так: (document, content). Я хочу преобразовать это в RDD как это: (document, word). Другими словами, я хочу сохранить информацию о документе в СДР.

Я использую RDD.flatMap(lambda (document, text): (document, re.split(' '), text)), чтобы разбить текст на слова, но не смог найти способ разбить последовательность слов на отдельные элементы.

Answer 1

In Scala кажется более интуитивным с case . В любом случае.

В pyspark:

%python
files = sc.wholeTextFiles("/FileStore/tables/x*.txt",0).map(lambda x: (x[0],x[1].replace('?',' ').replace('.',' ').replace('\r',' ').replace('\n',' ')  )) 
wordsAndFile = files.map(lambda k_v: (k_v[0], k_v[1].split(" ")))
res = wordsAndFile.map(lambda k_v: [(k_v[0], x) for x in k_v[1]])
final = res.flatMap(lambda x: x).filter(lambda x: x[1] is not u'')
final.collect()

Используя 3 файла, я получаю это, показанная часть: проще и меньше \ r, \ n вещей. Не применил replaceAll, просто чтобы продемонстрировать точку:

val files = sc.wholeTextFiles("/FileStore/tables/x*.txt",0) 
val lines2 = files.map { case (x, y) =>  (x, y.split(" ")) }
val lines3 = lines2.flatMap {  case (k, v) => { v.map(x => (k,x))    }  }
lines3.collect