Spark теряет 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

Я смог обойти это, сделав функцию полезности:

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)