scala - 无法在 Spark Sql 的 OutputMetrics 中获取 RecordsWritten

Question

我正在创建一个自定义的 spark 监听器并将其添加到运行的 sparkContext 但即使在这样做之后，如果我执行DataFrame 操作。

当我在下面运行时(非 DataFrame):

import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd}
    var recordsWrittenCount = 0L
    spark.sparkContext.addSparkListener(new SparkListener() {
      override def onTaskEnd(taskEnd: SparkListenerTaskEnd) {
        synchronized {
          recordsWrittenCount += taskEnd.taskMetrics.outputMetrics.recordsWritten
        }
      }
    })
    spark.sparkContext.parallelize(1 to 10, 2).saveAsTextFile(outputPath)
    println("Records Written: " + recordsWrittenCount)

然后我就可以得到recordsWritten

但是当我在下面运行时:

import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd}
    var recordsWrittenCount = 0L
    spark.sparkContext.addSparkListener(new SparkListener() {
      override def onTaskEnd(taskEnd: SparkListenerTaskEnd) {
        synchronized {
          recordsWrittenCount += taskEnd.taskMetrics.outputMetrics.recordsWritten
        }
      }
    })
    import spark.implicits._
    val someDF = Seq(
      (8, "bat"),
      (64, "mouse"),
      (-27, "horse")
    ).toDF("number", "word")
    someDF.write.save(outputPath)

    println("Records Written: " + recordsWrittenCount)

我得到了 0 条已写入的记录

谁能告诉我为什么会这样!!

附言:使用 Apache Spark 2.2

Answer 1

我想通了，因为它是 Apache Spark 2.2 中的已知错误，后来在 Apache Spark 2.3 中得到解决

https://issues.apache.org/jira/browse/SPARK-22605