Akumulator nie działa na klastrze działa lokalnie

Question

W oficjalnej dokumentacji zapłonowej, nie jest przykładem dla akumulatora, który jest używany w foreach rozmowy, który jest bezpośrednio na 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 

I wdrożone mojego własnego akumulatora:

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) 

W lokalnym ustawieniu działa to dobrze. Jednakże, jeśli uruchomię tę pracę w klastrze iskra wolnostojąca z kilku maszyn, robotnicy rzucać

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) 

na linii, która zwiększa akumulator myCounter.

Moje pytanie brzmi: czy akumulatory mogą być używane tylko w anonimowych funkcjach "najwyższego poziomu", które są stosowane bezpośrednio w RDD, a nie w zagnieżdżonych funkcjach? Jeśli tak, dlaczego moje połączenie udaje się lokalnie i kończy się niepowodzeniem w klastrze?

edycja: zwiększona szczegółowość wyjątku.

Answer 1

Co jeśli zdefiniować funkcję tak:

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

A potem nazwać tak:

foo(line, myCounter) 

?

Answer 2

Jeśli użyjesz "flatMap", wówczas "myCounter" nie będzie aktualizowany, ponieważ "flatMap" jest funkcją leniwą. Można użyć tego kodu:

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

Answer 3

W moim przypadku też akumulator był zerowy w zamknięciu, gdy użyłem „rozciąga App” aby utworzyć aplikację zapłonową jak pokazano poniżej

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

Wymieniłem rozciąga aplikacji z główną metodą() i pracowała w klastrze przędzy HDP 2,4 obiektów AccTest { zdecydowanie najważniejsze (argumenty: Tablica [String]): jednostka = {

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 

} }