Scala: удаление пустых значений массива из Spark DataFrame - PullRequest
Новая
       41

Scala: удаление пустых значений массива из Spark DataFrame

0 голосов
/ 11 ноября 2018

Я новый ученик Scala. Теперь дан DataFrame с именем df следующим образом: 

+-------+-------+-------+-------+
|Column1|Column2|Column3|Column4|
+-------+-------+-------+-------+
| [null]|  [0.0]|  [0.0]| [null]|
| [IND1]|  [5.0]|  [6.0]|    [A]|
| [IND2]|  [7.0]|  [8.0]|    [B]|
|     []|     []|     []|     []|
+-------+-------+-------+-------+

Я хотел бы удалить строки, если все столбцы - пустой массив (4-я строка).

Например, я мог бы ожидать, что результат будет: 

+-------+-------+-------+-------+
|Column1|Column2|Column3|Column4|
+-------+-------+-------+-------+
| [null]|  [0.0]|  [0.0]| [null]|
| [IND1]|  [5.0]|  [6.0]|    [A]|
| [IND2]|  [7.0]|  [8.0]|    [B]|
+-------+-------+-------+-------+

Я пытаюсь использовать isNotNull (например, val temp=df.filter(col("Column1").isNotNull && col("Column2").isNotNull && col("Column3").isNotNull && col("Column4").isNotNull).show() ), но все равно показывает все строки.

Я нашел решение на Python, использующее Hive UDF из link , но мне было нелегко пытаться преобразовать в действительный код scala. Я хотел бы использовать команду scala, похожую на следующий код: 

val query = "SELECT * FROM targetDf WHERE {0}".format(" AND ".join("SIZE({0}) > 0".format(c) for c in ["Column1", "Column2", "Column3","Column4"]))
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
sqlContext.sql(query)

Любая помощь будет оценена. Спасибо.

Ответы [ 2 ]

0 голосов
/ 12 ноября 2018

Другой подход (в дополнение к принятому ответу) будет использовать Наборы данных .
Например, имея класс case: 

case class MyClass(col1: Seq[String],
                   col2: Seq[Double],
                   col3: Seq[Double],
                   col4: Seq[String]) { 
    def isEmpty: Boolean = ...
}

Вы можете представить свой источник в виде типизированной структуры: 

import spark.implicits._ // needed to provide an implicit encoder/data mapper 

val originalSource: DataFrame = ... // provide your source
val source: Dataset[MyClass] = originalSource.as[MyClass] // convert/map it to Dataset

Таким образом, вы можете выполнить фильтрацию следующим образом: 

source.filter(element => !element.isEmpty) // calling class's instance method
0 голосов
/ 11 ноября 2018

Использование isNotNull или isNull не будет работать, потому что он ищет нулевое значение в DataFrame. Ваш пример DF не содержит нулевые значения, но пустые значения, есть разница.

Один вариант: вы можете создать новый столбец с длиной массива и фильтром, если массив равен нулю. 

  val dfFil = df
    .withColumn("arrayLengthColOne", size($"Column1"))
    .withColumn("arrayLengthColTwo", size($"Column2"))
    .withColumn("arrayLengthColThree", size($"Column3"))
    .withColumn("arrayLengthColFour", size($"Column4"))
    .filter($"arrayLengthColOne" =!= 0 && $"arrayLengthColTwo" =!= 0 
    && $"arrayLengthColThree" =!= 0 && $"arrayLengthColFour" =!= 0)
    .drop("arrayLengthColOne", "arrayLengthColTwo", "arrayLengthColThree", "arrayLengthColFour")

Оригинал DF: 

+-------+-------+-------+-------+
|Column1|Column2|Column3|Column4|
+-------+-------+-------+-------+
|    [A]|    [B]|    [C]|    [d]|
|     []|     []|     []|     []|
+-------+-------+-------+-------+

Новый DF: 

+-------+-------+-------+-------+
|Column1|Column2|Column3|Column4|
+-------+-------+-------+-------+
|    [A]|    [B]|    [C]|    [d]|
+-------+-------+-------+-------+

Вы также можете создать функцию, которая будет отображать все столбцы и делать это.

...