SparkでUDFの呼び出しを呼び出す

Question

Sparkを使用する1.6.1 UDFが呼び出された回数を呼び出したい。私は非常に高価なUDF（1コールあたり1秒）とを持っているので、私はこれをやりたいと思っています。私のスパークジョブを必要以上に遅くしてUDFをデータフレームのレコード数よりも頻繁に呼び出すと思われます。

この状況を再現できませんでしたが、UDFへの呼び出し回数が行数よりも多少異なるように見えるという簡単な例がありました。

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 

} 

アキュムレータを使用してUDFのカウントを呼び出す方法が適切でない場合、どうすればいいですか？

注：実際のSpark-Jobでは、実際のレコード数よりも約1.7倍高いコールカウントを取得します。

Answer 1

sparkアプリケーションでは、scala.Appを拡張する代わりにmain（）メソッドを定義する必要があります。 scala.Appのサブクラスが正しく動作しないことがあります。

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) 
     // [...] 
    } 
} 

これは問題を解決するはずです。