Как проверить все значения в столбцах, эффективно используя Spark?

Question

Мне интересно, как настроить динамический c фильтр по неизвестным столбцам в 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
    }

Мне интересно, есть ли лучший способ. Большое спасибо!

Answer 1

Почему бы просто не сделать df.filter + df.count, используя только API DataFrame?

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

val count = df.filter(filter_expr).count

//count: Long = 3

Answer 2

вы получили 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.toSeq получить Seq от Row. fromSeq restore Строка для возврата того же объекта.

Редактировать: для подсчета просто добавьте: .toSeq.filter(_ == "Red").size