Почему комбинатор никогда не вызывается в искре aggregateByKey?

Question

Я пытаюсь понять, как aggregateByKey работает в искре

В приведенном ниже примере преобразовывается

("David", 6), ("Abby", 4), ("David", 5), ("Abby", 5))

в

(Abby,Set(5, 4))
(David,Set(5, 6))

С кодом ниже

    val babyNamesCSV = spark.sparkContext.parallelize(List(("David", 6), ("Abby", 4), ("David", 5), ("Abby", 5)))

    babyNamesCSV.aggregateByKey(new HashSet[Int])(
        (k,v) => {
            println("start")
            println(k)
            println(v)
            println("end")
            k += v
        }, 
        (v,k) => {
            println("start2")
            println(k)
            println(v)
            println("end2")
            v ++ k
        }).map(line => {
            println(line)
            line
        }).take(100)

Я заметил, что сумматор println никогда не показывался на терминале sbt, хотя seqOp показывал, есть ли причина, почему?

Answer 1

Предполагая, что вы работаете в режиме local (не кластер / пряжа и т. Д.), Единственное, что я могу себе представить, это то, что babyNamesCSV имеет только 1 раздел, это может произойти, если у вас только 1 ядро ​​или вы установили spark.master=local[1].В этом случае объединитель никогда не вызывается, потому что никакие разделы не должны быть объединены ...

Попробуйте явно указать количество разделов:

val babyNamesCSV = spark.sparkContext.parallelize(List(("David", 6), ("Abby", 4), ("David", 5), ("Abby", 5)), numSlices = 2)

Answer 2

Почему бы вам не попробовать добавить третий элемент ввода с одной из клавиш в ваших данных.Затем найдите printlns из обеих функций.

Причина может заключаться в том, что работники / исполнители, которые не находятся на том же компьютере / jvm, что и драйвер, не могут показать свой стандартный вывод вашей программе драйвера.Надеюсь, это поможет.