累加器在集群上失败，在本地工作答案

【问题标题】：Accumulator fails on cluster, works locally累加器在集群上失败，在本地工作
【发布时间】：2013-07-21 14:33:14
【问题描述】：

在官方 spark 文档中，有一个累加器示例，用于直接在 RDD 上的 foreach 调用：

scala> val accum = sc.accumulator(0)
accum: spark.Accumulator[Int] = 0

scala> sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum += x)
...
10/09/29 18:41:08 INFO SparkContext: Tasks finished in 0.317106 s

scala> accum.value
res2: Int = 10

我实现了自己的累加器：

val myCounter = sc.accumulator(0)

val myRDD = sc.textFile(inputpath) // :spark.RDD[String]

myRDD.flatMap(line => foo(line)) // line 69

def foo(line: String) = {
   myCounter += 1  // line 82 throwing NullPointerException
   // compute something on the input
}
println(myCounter.value)

在本地环境中，这工作得很好。但是，如果我在具有多台机器的 spark 独立集群上运行此作业，工作人员会抛出一个

13/07/22 21:56:09 ERROR executor.Executor: Exception in task ID 247
java.lang.NullPointerException
    at MyClass$.foo(MyClass.scala:82)
    at MyClass$$anonfun$2.apply(MyClass.scala:67)
    at MyClass$$anonfun$2.apply(MyClass.scala:67)
    at scala.collection.Iterator$$anon$21.hasNext(Iterator.scala:440)
    at scala.collection.Iterator$$anon$19.hasNext(Iterator.scala:400)
    at spark.PairRDDFunctions.writeToFile$1(PairRDDFunctions.scala:630)
    at spark.PairRDDFunctions$$anonfun$saveAsHadoopDataset$2.apply(PairRDDFunctions.scala:640)
    at spark.PairRDDFunctions$$anonfun$saveAsHadoopDataset$2.apply(PairRDDFunctions.scala:640)
    at spark.scheduler.ResultTask.run(ResultTask.scala:77)
    at spark.executor.Executor$TaskRunner.run(Executor.scala:98)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:722)

在累加器myCounter递增的那一行。

我的问题是：累加器只能用于直接应用于 RDD 而不是嵌套函数的“顶级”匿名函数吗？如果是，为什么我的调用在本地成功而在集群上失败？

编辑：增加异常的详细程度。

【问题讨论】：

你能发布更多工人的回溯吗？
你试过sc.broadcast(myCounter)吗？
broadcast 不返回只读值吗？来自official API docs：“将只读变量广播到集群，返回一个Broadcast对象用于在分布式函数中读取它。该变量将只发送到每个集群一次。”
好点。无论如何，看起来 spark 没有将您的累加器发送到集群，因此是空指针。不知道如何解决这个问题，除了稍微改变你的算法。
您不应该将 sc.broadcast() 与累加器一起使用。

标签： scala mapreduce apache-spark

【解决方案1】：

在我的情况下，当我使用“扩展应用程序”创建如下所示的 spark 应用程序时，累加器在关闭时为空

    object AccTest extends App {


    val conf = new SparkConf().setAppName("AccTest").setMaster("yarn-client")
    val sc = new SparkContext(conf)
    sc.setLogLevel("ERROR")

    val accum = sc.accumulator(0, "My Accumulator")
    sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum += x)

    println("count:" + accum.value)

    sc.stop
  }
}

我用 main() 方法替换了 extends App，它在 HDP 2.4 的 YARN 集群中工作

object AccTest {

    def main(args: Array[String]): Unit = {

        val conf = new SparkConf().setAppName("AccTest").setMaster("yarn-client")
        val sc = new SparkContext(conf)
        sc.setLogLevel("ERROR")

        val accum = sc.accumulator(0, "My Accumulator")
        sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum += x)

        println("count:" + accum.value)

        sc.stop
    }
}

工作

【讨论】：

谢谢！这节省了我 2 天的调试时间。有谁知道为什么 scala.App 在这里不起作用？

【解决方案2】：

如果你这样定义函数会怎样：

def foo(line: String, myc: org.apache.spark.Accumulator[Int]) = {
    myc += 1
}

然后这样称呼它：

foo(line, myCounter)

【讨论】：

这似乎是正确的，您可以将您创建的累加器传递给方法

【解决方案3】：

如果您使用“flatMap”，那么“myCounter”将不会更新，因为“flatMap”是惰性函数。您可以使用此代码：

myRDD.foreach(line => foo(line))
def foo(line: String) = {myCounter +=1}
println(myCounter.value)

【讨论】：