Spark: subtractByKey issue (pyspark)

Question

У меня есть некоторые проблемы с subtractByKey.

У меня есть 2 файла: Первый один, как: (Client ID + почтовый клиент)

client_id emails 

4A85FD8E-197D-2AE3-B939-A527AFF16A04 imperdiet.non.vestibulum@mon***tur.com 
D48D530C-CF68-DAF1-18F0-E0A0A03F3E06 rutrum.urna@estm***ncus.net:facilisis@i****m.ca 
40815230-25DC-9EA0-01D1-2706B4B56958 iaculis.nec.eleifend@gr****nc.net 
... 

и второй один: (только почта)

pharetra@P****s.com 
ut.aliquam@o****m.org 
erat@a****e.edu 
.... 

Некоторые строки в первом файле могут иметь 2 (или более) почты в этом формате:

mail:mail 

Что я сделал:

*test1=sc.textFile("file1") 
*test2=sc.textFile("file2") 
*test3=test1.subtractByKey(test2) 

и результат ...:

[(u'A', u'B'), (u'A', u'D'), (u'A', u'1'), (u'A', u'D'), (u'A', u'D'), (u'A', u'B'), (u'A', u'F'), (u'A', u'E'), (u'A', u'9'), (u'A', u'5'), (u'A', u'9'), (u'A', u'6'), (u'c', u'l'), (u'E', u'8'), (u'E', u'4'), (u'E', u'6'), (u'E', u'6'), (u'E', u'7'), (u'E', u'5'), (u'E', u'5'), (u'E', u'5'), (u'E', u'2'), (u'E', u'8'), (u'C', u'2'), (u'C', u'5'), (u'C', u'6'), (u'C', u'C'), (u'C', u'E'), (u'C', u'3'), (u'C', u'F'), (u'C', u'4'), (u'C', u'B'), (u'C', u'F'), (u'C', u'F'), (u'C', u'8'), (u'C', u'0'), (u'1', u'D'), (u'1', u'2'), (u'1', u'3'), (u'1', u'8'), (u'1', u'0'), (u'1', u'F'), ... ] 

Я хотел удалить клиентов в первом файле, который имел свою почту во втором файле, но это не работает.

Answer 1

примечание: Я не очень хорошо знаком с pyspark но искра апи должна быть же.

сначала вы должны сделать по электронной почте в качестве ключевого

rdd1=sc.textFile("file1").map(lambda line: (line.split(" ")[0], line.split(" ")[1]))

это даст вам RDD из

[(4A85FD8E-197D-2AE3-B939-A527AFF16A04,imperdiet.non.vestibulum@mon***tur.com)] 

тогда как может быть много электронной почты, вы должны сделать flatMapValues()

rdd2 = rdd1.flatMapValues(lambda email: email.split(":"))

это даст вам пару RDD и каждый из них содержит только одну электронную почту

теперь вы можете переключить ключ и значение

rdd3=rdd2.map(lambda kv: (kv[1], kv[0])) 

Теперь вы получаете RDD, что с помощью электронной почты пользователя в качестве ключа и UUID в качестве значения , таких как

[(imperdiet.non.vestibulum@mon***tur.com, 4A85FD8E-197D-2AE3-B939-A527AFF16A04)] 

теперь вы должны найти какой адрес электронной почты UUID содержатся в file2, чтобы сделать это, необходимо загрузить второй файл в качестве РДА:

secondRdd = sc.textFile("file2").map(lambda line: (line, 1)) 

и выполните join и отредактируйте результат соединения rdd.

rdd4 = rdd3.join(secondRdd).map(lambda kv: (kv[1][0], kv[0])) 

если все это прямо сейчас, вы должны получить RDD, который как формат (UUID, email), который представляет всех пользователей, чьи сообщения электронной почты происходит в file2,

, то вы можете сделать subtractByKey() с rdd1 который мы первоначально получил.