Pyspark - как отфильтровать RDD с помощью Broadcast Dictionary по значению ключа И - PullRequest
Новая
       13

Pyspark - как отфильтровать RDD с помощью Broadcast Dictionary по значению ключа И

1 голос
/ 22 апреля 2019

Я пытаюсь отфильтровать большой RDD на основе широковещательной переменной.

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

nested_filter = {"india":'ind',"usa":'us'}
b_filter = sc.broadcast(nested_filter)


rdd_set = sc.parallelize([('india','ind'),('india', 'nope') , ('usa','us'),
                ('japan','jpn'),('uruguay','urg')])

Фильтр: 

rdd_set.filter(lambda fields: fields[0] in b_filter.value).collect()

Возвращает: 

[('india', 'ind'), ('india', 'nope'), ('usa', 'us')]

Моя проблема заключается в том, что я хочу отфильтровать ключи в словаре рассылки, а также значение, связанное с ключом.

Правильный результат должен быть следующим: 

[('india', 'ind'), ('usa', 'us')]

Фактический СДР будет иметь несколько миллиардов строк строк с широковещательным словарем, содержащим несколько миллионов ключей.Может кто-нибудь показать мне наиболее эффективный способ сделать это?

1 Ответ

1 голос
/ 22 апреля 2019

Вы можете использовать items(), чтобы получить список кортежей ключ-значение из вашего словаря.Затем проверьте, есть ли строка в этом списке: 

rdd_set.filter(lambda fields: fields in b_filter.value.items()).collect()
#[('india', 'ind'), ('usa', 'us')]
...