火花在标准输出上丢失 println()答案

【问题标题】：Spark losing println() on stdout火花在标准输出上丢失 println()
【发布时间】：2016-01-18 11:36:17
【问题描述】：

我有以下代码：

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() 输出！我在这里错过了什么吗？谢谢！

【问题讨论】：

标签： scala apache-spark println accumulator

【解决方案1】：

这是一个概念性问题...

想象你有一个大集群，由许多工作人员组成，比如说n工作人员，这些工作人员存储RDD或DataFrame的分区，想象你在这些数据中启动一个map任务，并在内部map 你有一个print 声明，首先：

这些数据将打印在哪里？
哪个节点有优先级，哪个分区？
如果所有节点都并行运行，谁会先打印？
如何创建此打印队列？

这些问题太多了，因此apache-spark 的设计者/维护者在逻辑上决定放弃对任何map-reduce 操作中的print 语句的任何支持（这包括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)

【讨论】：

我相信 println 工作得很好：它只是转到运行 spark 执行器的计算机上的 stdout /stderr。因此，除非您有办法捕获这些日志中的内容，否则您将永远看不到它。但是，如果您使用纱线，则有一个命令可以为您打印出来。
当论证有效时，Spark 不会执行任何类型的静态分析来删除代码。正如@David 所解释的那样，输出不会发送给驱动程序STDOUT

【解决方案2】：

我可以通过创建一个实用函数来解决它：

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

【讨论】：

因为 Spark 认为它是在调用 Utility 函数而不是调用 print 函数。 Spark 显然没有（实际上也不能）检查其实用函数中的每一行。
你正在做的是在你的驱动程序中实例化一个对象。如果没有确切的确切模型，我不会指望这种行为。预期行为会随着程序的任何更改或调用 PrintUtility 对象的方式发生不可预测的变化。如果你想收集日志，就用标准的方法去做，不要发明你不理解的随机机制。你对它为什么起作用的解释是危险的错误——没有禁止做你所做的事情；没有代码检查器可以确保您不作弊：所有行为都遵循系统设计