Аккумулятор выходит из строя на кластере, работает локально

В официальной документации по искрам есть пример для аккумулятора, который используется в вызове foreach, который находится непосредственно на RDD:

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)

В локальной настройке это работает отлично. Однако, если я выполняю эту работу на искровом автономном кластере с несколькими машинами, рабочие бросают

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, а не к вложенным функциям? Если да, то почему мой вызов выполняется локально и сбой в кластере?

изменить: увеличенная многословность исключения.

Ответ 1

В моем случае аккумулятор также был пустым в закрытии, когда я использовал 'extends App' для создания искрового приложения, как показано ниже

    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
  }
}

Я заменил приложение extends на метод main() и работал в кластере YARN в HDP 2.4 объект 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

Что делать, если вы определяете функцию следующим образом:

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

И затем назовите его следующим образом:

foo(line, myCounter)

?

Ответ 3

Если вы используете "flatMap", тогда "myCounter" не будет обновляться, потому что "flatMap" - это функция lazzy. Вы можете использовать этот код:

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