Спарк теряя Println() на стандартный вывод

Question

У меня есть следующий код:

val blueCount = sc.accumulator[Long](0) 
val output = input.map { data => 
    for (value <- data.getValues()) { 
    if (record.getEnum() == DataEnum.BLUE) { 
     blueCount += 1 
     println("Enum = BLUE : " + value.toString() 
    } 
    } 
    data 
}.persist(StorageLevel.MEMORY_ONLY_SER) 

output.saveAsTextFile("myOutput") 

---

Тогда blueCount не равна нулю, но у меня нет Println() выход! Я что-то пропустил? Благодаря!

Answer 1

Я был в состоянии работать вокруг, сделав UtilityFunction:

object PrintUtiltity { 
    def print(data:String) = { 
     println(data) 
    } 
} 

Answer 2

Это концептуальный вопрос ...

Представьте у вас есть большой кластер, состоящий из многих рабочих скажут n рабочих и те работники хранят переборку RDD или DataFrame, представьте Вы начинаете map задачу по что данные, и внутри этого map у вас есть print заявление, прежде всего:

• Где эти данные распечатываются?
• Какой узел имеет приоритет и какой раздел?
• Если все узлы работают параллельно, кто будет печататься первым?
• Как будет создана очередь печати?

Это слишком много вопросов, таким образом, конструкторы/Сопровождающие из apache-spark решил логически отказаться от любой поддержки print операторов внутри любой map-reduce операции (это включает accumulators и даже broadcast переменные).

Это также имеет смысл, потому что Spark - это язык , рассчитанный на для очень больших наборов данных. Хотя печать может быть полезна для тестирования и отладки, вы не захотите печатать каждую строку DataFrame или RDD, потому что они созданы, чтобы иметь миллионы или миллиарды строк! Итак, зачем решать эти сложные вопросы, когда вы даже не хотите печатать в первую очередь?

Чтобы доказать это, вы можете запустить этот SCALA код, например:

// Let's create a simple RDD 
val rdd = sc.parallelize(1 to 10000) 

def printStuff(x:Int):Int = { 
    println(x) 
    x + 1 
} 

// It doesn't print anything! because of a logic design limitation! 
rdd.map(printStuff) 

// But you can print the RDD by doing the following: 
rdd.take(10).foreach(println)