Spark Scala：如何同时过滤 RDD 和更新计数器答案

【问题标题】：Spark Scala: How to filter RDD and update counter at the same timeSpark Scala：如何同时过滤 RDD 和更新计数器
【发布时间】：2018-12-18 09:46:46
【问题描述】：

我是 spark/scala 的新手。我最初的 RDD 是 Records 类型，记录的布局是：

a_key, b_key,c_key,f_name,l_name,address

现在我必须：

删除 a_key 或 b_key 或 c_key 为 null/空的记录
我必须同时更新无效记录的计数器。

我试过这样：

sc.register( recordStatsAccumulator, "Stat accumulator for " + filename )

val nullFilteredRecords = records.map{ record =>

  if( record.A_KEY.isEmpty ||
    record.B_KEY.isEmpty ||
    record.C_KEY.isEmpty )
  {
    recordStatsAccumulator.add( ValidationLoggingUtil.INVALID )
  }

  record

 }
 .filter( record =>
    !record.A_KEY.isEmpty &&
      !record.B_KEY.isEmpty &&
      !record.C_KEY.isEmpty
  )

但是，此代码效率不高，因为它两次遍历整个 RDD。首先，更新无效记录的计数器，然后再次删除无效记录。

有没有更好/有效的方法来做到这一点？

【问题讨论】：

标签： scala apache-spark apache-spark-sql

【解决方案1】：

我认为您可以一步合并这两个操作。像这样：

val nullFilteredRecords = records.filter { record =>
  if( record.A_KEY.isEmpty ||
    record.B_KEY.isEmpty ||
    record.C_KEY.isEmpty ) {
    recordStatsAccumulator.add( ValidationLoggingUtil.INVALID )
  }
  !record.A_KEY.isEmpty && !record.B_KEY.isEmpty && !record.C_KEY.isEmpty
}

【讨论】：