Scala - Какой самый эффективный способ получить имена пустых столбцов из фрейма данных spark?

Question

У меня есть df как это:

+----------+----------+----------+----------+----------+----------+----------+
|   user_id|     apple|    orange|    banana|      pear|     table|      desk|
+----------+----------+----------+----------+----------+----------+----------+
|         1|        13|      null|        55|      null|      null|      null|
|         2|        30|      null|      null|      null|      null|      null|
|         3|      null|      null|        50|      null|      null|      null|
|         4|         1|      null|         3|      null|      null|      null|
+----------+----------+----------+----------+----------+----------+----------+

Я хотел бы получить массив [String], который содержит имена столбцов фруктов, которые имеют только нулевые значения. Я хотел бы сделать это на очень большом кадре данных, поэтому я не хочу суммировать столбцы, мне нужен более быстрый и более эффективный способ. Мне нужен код Scala.

Так что мне нужен этот список:

List(orange,pear)

У меня есть это решение сейчас, суммирование столбцов, но мне нужно решение без суммирования всех столбцов:

val fruitList:  Array[String] = here are the fruit names 
val nullFruits: Array[String] = fruitList.filter(col => dataFrame.agg(sum(col)).first.get(0) == null)

Answer 1

Вы можете добиться этого, используя Spark's describe тоже:

val r1  = df.select(fruitList.head, fruitList.tail :_*)
  .summary("count")

//alternatively

val r1 = df.select(fruitList.head, fruitList.tail :_*)
   .describe()
   .filter($"summary" === "count")

+-------+-----+------+------+----+
|summary|apple|orange|banana|pear|
+-------+-----+------+------+----+
|  count|    3|     0|     3|   0|
+-------+-----+------+------+----+

И для извлечения желаемых значений:

r1.columns.tail
  .map(c => (c,r1.select(c).head.getString(0) == "0"))
  .filter(_._2 == true)
  .map(_._1)

, что дает:

Array(orange, pear)

Answer 2

Можно использовать функцию уменьшение набора данных.

Сначала мы определим функцию уменьшения, которая сравнивает попарно две строки. Результат для каждого столбца строк: либо null (если значения в обеих строках для соответствующего столбца null), либо 1 в противном случае.

def reduceNull(r1: Row, r2: Row): Row = {
  Row.fromSeq(
    for( i <- 0 to r1.length-1) yield if( r1.isNullAt(i) && r2.isNullAt(i) ) null else 1
    )
}

Чтобы уменьшить объем данных В процессе обработки мы выбираем только те столбцы, в которых мы заинтересованы, а затем применяем функцию сокращения:

val fruitList = Seq("apple", "orange", "banana", "pear")
val resultRow = df.select(fruitList.head, fruitList.tail:_*)
  .reduce(reduceNull _)

Результатом функции уменьшения является отдельная строка, содержащая в каждом столбце либо null, либо * 1015. *, в зависимости от всех значений в этом столбце в исходном фрейме данных.

val colsWithAllRowsNull = (for( i <- 0 to resultRow.size-1) yield (fruitList(i), resultRow.get(i)))
  .filter(e => e._2 == null)
  .map(_._1)
  .toList
println(colsWithAllRowsNull)

печатает

List(orange, pear)

EDIT : Оказалось, что для моей установки (Spark Standalone, 4 ядра, 2 линии данных Mio) использование Dataset.describe занимает только половину времени функции уменьшения.