scala - 如何获取写入的记录数(使用DataFrameWriter的save操作)？

Question

使用spark保存记录时，有没有办法获取写入的记录数？虽然我知道它目前不在规范中，但我希望能够执行以下操作:

val count = df.write.csv(path)

或者，能够对步骤的结果进行内联计数(最好不只使用标准累加器)将(几乎)同样有效。 IE。:

dataset.countTo(count_var).filter({function}).countTo(filtered_count_var).collect()

有任何想法吗？

Answer 1

我会用 SparkListener 可以拦截onTaskEnd或 onStageCompleted可用于访问任务指标的事件。

任务指标为您提供 Spark 用于在 SQL 选项卡(在查询的详细信息中)显示指标的累加器。



例如，以下查询:

spark.
  read.
  option("header", true).
  csv("../datasets/people.csv").
  limit(10).
  write.
  csv("people")

正好给出 10 个输出行，所以 Spark 知道它(你也可以)。



您还可以探索 Spark SQL 的 QueryExecutionListener :

The interface of query execution listener that can be used to analyze execution metrics.

您可以注册 QueryExecutionListener使用 ExecutionListenerManager 可用作 spark.listenerManager .

scala> :type spark.listenerManager
org.apache.spark.sql.util.ExecutionListenerManager

scala> spark.listenerManager.
clear   clone   register   unregister

我认为它更接近“裸机”，但之前没有使用过。

@D3V (在评论部分)提到访问 numOutputRows SQL 指标使用 QueryExecution结构化查询。值得考虑的东西。

scala> :type q
org.apache.spark.sql.DataFrame

scala> :type q.queryExecution.executedPlan.metrics
Map[String,org.apache.spark.sql.execution.metric.SQLMetric]

q.queryExecution.executedPlan.metrics("numOutputRows").value