Поиск значений в широковещательной переменной

Question

Я хочу присоединиться к двум наборам, применяя широковещательную переменную. Я пытаюсь реализовать первое предложение от Spark: what's the best strategy for joining a 2-tuple-key RDD with single-key RDD?

val emp_newBC = sc.broadcast(emp_new.collectAsMap()) 
val joined = emp.mapPartitions({ iter => 
     val m = emp_newBC.value 
     for { 
     ((t, w)) <- iter 
     if m.contains(t) 
     } yield ((w + '-' + m.get(t).get),1) 
    }, preservesPartitioning = true) 

Однако, как указано здесь: broadcast variable fails to take all data мне нужно использовать сбор(), а не collectAsMAp(). Я попытался скорректировать свой код, как показано ниже:

val emp_newBC = sc.broadcast(emp_new.collect()) 
val joined = emp.mapPartitions({ iter => 
     val m = emp_newBC.value 
     for { 
     ((t, w)) <- iter 
     if m.contains(t) 
     amk = m.indexOf(t) 
     } yield ((w + '-' + emp_newBC.value(amk)),1) //yield ((t, w), (m.get(t).get)) //((w + '-' + m.get(t).get),1) 
    }, preservesPartitioning = true) 

Но, похоже, m.contains (t) не отвечает. Как я могу это исправить?

Заранее спасибо.

Answer 1

Как насчет чего-то подобного?

val emp_newBC = sc.broadcast(emp_new.groupByKey.collectAsMap) 

val joined = emp.mapPartitions(iter => for { 
    (k, v1) <- iter 
    v2 <- emp_newBC.value.getOrElse(k, Iterable()) 
} yield (s"$v1-$v2", 1)) 

Что касается вашего кода ... Насколько я понимаю emp_new является RDD[(String, String)]. Когда он будет собран, вы получите Array[(String, String)]. При использовании

((t, w)) <- iter 

t является String так m.contains(t) всегда будет возвращать false.

Другая проблема, которую я вижу, это preservesPartitioning = true внутри mapPartitions. Там несколько возможных сценариев:

1. emp разбивается, и вы хотите joined быть разбит на разделы, а также. Так как вы меняете ключ от t до некоторого нового значения, разбиение не может быть сохранено, и итоговое значение RDD должно быть переразделено. Если вы используете preservesPartitioning = true, вывод RDD закончится неправильными разделами.
2. emp разделен, но вам не нужно разбиение на разделы для joined. Нет причин использовать preservesPartitioning.
3. emp не разделен. Значение preservesPartitioning не влияет.