Anzahl Aufrufe von UDF in Spark

Question

Verwenden von Spark 1.6.1 Ich möchte die Anzahl der Aufrufe einer UDF aufrufen. Ich möchte dies tun, weil ich eine sehr teure UDF (~ 1sec pro Anruf) und habe Ich vermute, dass die UDF häufiger als die Anzahl der Datensätze in meinem Datenframe aufgerufen wird, so dass mein Funke Job langsamer als notwendig.

Obwohl ich diese Situation nicht reproduzieren konnte, kam ich zu einem einfachen Beispiel, das zeigt, dass die Anzahl der Aufrufe an die UDF anders zu sein scheint (hier: weniger) als die Anzahl der Zeilen, wie kann das sein?

import org.apache.spark.sql.SQLContext 
import org.apache.spark.{SparkConf, SparkContext} 
import org.apache.spark.sql.functions.udf 

object Demo extends App { 
    val conf = new SparkConf().setMaster("local[4]").setAppName("Demo") 
    val sc = new SparkContext(conf) 
    sc.setLogLevel("WARN") 
    val sqlContext = new SQLContext(sc) 
    import sqlContext.implicits._ 


    val callCounter = sc.accumulator(0) 

    val df= sc.parallelize(1 to 10000,numSlices = 100).toDF("value") 

    println(df.count) // gives 10000 

    val myudf = udf((d:Int) => {callCounter.add(1);d}) 

    val res = df.withColumn("result",myudf($"value")).cache 

    println(res.select($"result").collect().size) // gives 10000 
    println(callCounter.value) // gives 9941 

} 

Wenn die Verwendung eines Akkumulators nicht der richtige Weg ist, die Zählerstände der UDF aufzurufen, wie sonst könnte ich das tun?

Hinweis: In meiner aktuellen Spark-Job, erhalten Sie eine Anruf-Zählung, die etwa 1,7 mal höher ist als die tatsächliche Anzahl der Datensätze.

Answer 1

Spark-Anwendungen sollten eine main() - Methode definieren, anstatt scala.App zu erweitern. Unterklassen von scala.App funktionieren möglicherweise nicht richtig.

import org.apache.spark.sql.SQLContext 
import org.apache.spark.{SparkConf, SparkContext} 
import org.apache.spark.sql.functions.udf 

object Demo extends App { 
    def main(args: Array[String]): Unit = { 
     val conf = new SparkConf().setAppName("Simple Application").setMaster("local[4]") 
     val sc = new SparkContext(conf) 
     // [...] 
    } 
} 

Dies sollte Ihr Problem lösen.