如何使用 Spark 有效地检查列中的所有值？答案

【问题标题】：How to check all values in columns efficiently using Spark?如何使用 Spark 有效地检查列中的所有值？
【发布时间】：2020-06-18 06:50:23
【问题描述】：

我想知道如何根据 Spark 中的未知列进行动态过滤。

例如，数据框如下：

+-------+-------+-------+-------+-------+-------+
| colA  | colB  |  colC |  colD |  colE |  colF | 
+-------+-------+-------+-------+-------+-------+
| Red   | Red   | Red   | Red   | Red   | Red   |
| Red   | Red   | Red   | Red   | Red   | Red   |
| Red   | Blue  | Red   | Red   | Red   | Red   |
| Red   | Red   | Red   | Red   | Red   | Red   |
| Red   | Red   | Red   | Red   | Blue  | Red   |
| Red   | Red   | White | Red   | Red   | Red   |
+-------+-------+-------+-------+-------+-------+

这些列只能在运行时知道，这意味着它可以有 colG、H .. 我需要检查整个列的值是否为红色，然后得到一个计数，在上述情况下为 3，因为 colA、colD 和 ColF 列都是红色的。

我正在做的事情如下所示，而且速度很慢..

   val allColumns = df.columns
   df.foldLeft(allColumns) {

      (df, column) =>
        val tmpDf = df.filter(df(column) === "Red")
        if (tmpDf.rdd.isEmpty) {
          count += 1
        }
        df
    }

我想知道是否有更好的方法。非常感谢！

【问题讨论】：

标签： scala apache-spark

【解决方案1】：

您进行了 N RDD 扫描，其中 N 是列数。您可以一次扫描所有这些并并行减少。例如这种方式：

df.reduce((a, r) => Row.fromSeq(a.toSeq.zip(r.toSeq)
    .map { case (a, r) => 
          if (a == "Red" && r == "Red") "Red" else "Not" 
    }
))

res11: org.apache.spark.sql.Row = [Red,Not,Not]

此代码将执行一次 RDD 扫描，然后在 reduce 中迭代 Row 列。 Row.toSeq 从 Row 获取 Seq。 fromSeq restore Row 返回相同的对象。

编辑：计数只需添加：.toSeq.filter(_ == "Red").size

【讨论】：

【解决方案2】：

为什么不简单地使用仅使用 DataFrame API 的 df.filter + df.count？

val filter_expr = df.columns.map(c => col(c) === lit("Red")).reduce(_ and _)

val count = df.filter(filter_expr).count

//count: Long = 3

【讨论】：