计算数据框 SPARK 上的过滤项目

Question

我有以下数据框：df

在某些时候，我需要根据时间戳（毫秒）过滤掉项目。 但是，保存过滤了多少记录对我来说很重要（如果太多，我想让工作失败） 天真地我能做到：

======Lots of calculations on df ======
val df_filtered = df.filter($"ts" >= startDay && $"ts"  <= endDay)
val filtered_count = df.count - df_filtered.count

但感觉完全是矫枉过正，因为 SPARK 将执行整个执行树 3 次（过滤和 2 个计数）。 Hadoop MapReduce 中的这项任务非常简单，因为我可以为过滤的每一行维护计数器。 有没有更有效的方法，只能找到累加器却无法连接过滤器。

建议的方法是在过滤器之前缓存 df，但是由于 DF 大小，我更喜欢这个选项作为最后的手段。

Answer 1

Spark 1.6.0 代码：

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

object Main {

  val conf = new SparkConf().setAppName("myapp").setMaster("local[*]")
  val sc = new SparkContext(conf)
  val sqlContext = new SQLContext(sc)

  case class xxx(a: Int, b: Int)

  def main(args: Array[String]): Unit = {

    val df = sqlContext.createDataFrame(sc.parallelize(Seq(xxx(1, 1), xxx(2, 2), xxx(3,3))))

    val acc = sc.accumulator[Long](0)

    val filteredRdd = df.rdd.filter(r => {
      if (r.getAs[Int]("a") > 2) {
        true
      } else {
        acc.add(1)
        false
      }
    })

    val filteredRddDf = sqlContext.createDataFrame(filteredRdd, df.schema)

    filteredRddDf.show()

    println(acc.value)
  }
}

Spark 2.x.x 代码：

import org.apache.spark.sql.SparkSession

object Main {

  val ss = SparkSession.builder().master("local[*]").getOrCreate()
  val sc = ss.sparkContext

  case class xxx(a: Int, b: Int)

  def main(args: Array[String]): Unit = {

    val df = ss.createDataFrame(sc.parallelize(Seq(xxx(1, 1), xxx(2, 2), xxx(3,3))))

    val acc = sc.longAccumulator

    val filteredDf = df.filter(r => {
      if (r.getAs[Int]("a") > 2) {
        true
      } else {
        acc.add(1)
        false
      }
    }).toDF()


    filteredDf.show()

    println(acc.value)

  }
}