スパーク：書いた行の数を取得するには？

Question

Sparkの保存操作で書き込まれる行の数を知る方法があるのだろうかと思います。私は、それを書く前にRDDを数えるだけで十分だと知っていますが、それをしないと同じ情報を持つ方法があるかどうかを知りたいと思います。

は、 マルコ

Answer 1

を行うことができますあなたが本当に望むのであれば、カスタムリスナーを追加し、書いた行の数をから抽出することができます。非常に単純な例は、次のようになります。

import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd} 

var recordsWrittenCount = 0L 

sc.addSparkListener(new SparkListener() { 
    override def onTaskEnd(taskEnd: SparkListenerTaskEnd) { 
    synchronized { 
     recordsWrittenCount += taskEnd.taskMetrics.outputMetrics.recordsWritten 
    } 
    } 
}) 

sc.parallelize(1 to 10, 2).saveAsTextFile("/tmp/foobar") 
recordsWrittenCount 
// Long = 10 

この部分は、内部使用を目的としています。

Answer 2

（様々なコメントで明示的に作ったとして）受け入れ答えは、より密接OPは特定のニーズに合った、ありがとうございます、それにもかかわらず、この答えが過半数を合わせます。

最も効率的なアプローチは、アキュムレータを使用することです：http://spark.apache.org/docs/latest/programming-guide.html#accumulators

val accum = sc.accumulator(0L) 

data.map { x => 
    accum += 1 
    x 
} 
.saveAsTextFile(path) 

val count = accum.value 

あなたはその後、便利なポン引きでこれをラップすることができます：

implicit class PimpedStringRDD(rdd: RDD[String]) { 
    def saveAsTextFileAndCount(p: String): Long = { 
    val accum = rdd.sparkContext.accumulator(0L) 

    rdd.map { x => 
     accum += 1 
     x 
    } 
    .saveAsTextFile(p) 

    accum.value 
    } 
} 

ですから、

val count = data.saveAsTextFileAndCount(path) 

Answer 3

あなたは

taskEnd.taskInfo.accumulables 

を見ればあなたはそれが順番にListBufferでAccumulableInfoを以下にバンドルされていることがわかります。あなたは明らかに出力行の数がlistBufferの7位にあるので、正しい方法は、行がカウントは我々が行が書かれて取得することができ

taskEnd.taskInfo.accumulables(6).value.get 

で書かれて取得する見ることができます

AccumulableInfo(1,Some(internal.metrics.executorDeserializeTime),Some(33),Some(33),true,true,None), 
AccumulableInfo(2,Some(internal.metrics.executorDeserializeCpuTime),Some(32067956),Some(32067956),true,true,None), AccumulableInfo(3,Some(internal.metrics.executorRunTime),Some(325),Some(325),true,true,None), 
AccumulableInfo(4,Some(internal.metrics.executorCpuTime),Some(320581946),Some(320581946),true,true,None), 
AccumulableInfo(5,Some(internal.metrics.resultSize),Some(1459),Some(1459),true,true,None), 
AccumulableInfo(7,Some(internal.metrics.resultSerializationTime),Some(1),Some(1),true,true,None), 
AccumulableInfo(0,Some(number of output rows),Some(3),Some(3),true,true,Some(sql) 

以下の方法により、（私はちょうどzero323の答え@に変更）

import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd} 

var recordsWrittenCount = 0L 

sc.addSparkListener(new SparkListener() { 
    override def onTaskEnd(taskEnd: SparkListenerTaskEnd) { 
    synchronized { 
     recordsWrittenCount += taskEnd.taskInfo.accumulables(6).value.get.asInstanceOf[Long] 
    } 
    } 
}) 

sc.parallelize(1 to 10, 2).saveAsTextFile("/tmp/foobar") 
recordsWrittenCount