Vfrdtyky

I'm a new learner of Scala. Now given a DataFrame named df as follows:

+-------+-------+-------+-------+

|Column1|Column2|Column3|Column4|

+-------+-------+-------+-------+

| [null]|  [0.0]|  [0.0]| [null]|

| [IND1]|  [5.0]|  [6.0]|    [A]|

| [IND2]|  [7.0]|  [8.0]|    [B]|

|     |     |     |     |

+-------+-------+-------+-------+

I'd like to delete rows if all columns is an empty array (4th row).

For example I might expect the result is:

+-------+-------+-------+-------+

|Column1|Column2|Column3|Column4|

+-------+-------+-------+-------+

| [null]|  [0.0]|  [0.0]| [null]|

| [IND1]|  [5.0]|  [6.0]|    [A]|

| [IND2]|  [7.0]|  [8.0]|    [B]|

+-------+-------+-------+-------+

I'm trying to use isNotNull (like val temp=df.filter(col("Column1").isNotNull && col("Column2").isNotNull && col("Column3").isNotNull && col("Column4").isNotNull).show()
) but still show all rows.

I found python solution of using a Hive UDF from link, but I had hard time trying to convert to a valid scala code. I would like use scala command similar to the following code:

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)

Any help would be appreciated. Thank you.

Using the isNotNull or isNull will not work because it is looking for a 'null' value in the DataFrame. Your example DF does not contain null values but empty values, there is a difference there.

One option: You could create a new column that has the length of of the array and filter for if the array is zero.

  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")

Original DF:

+-------+-------+-------+-------+

|Column1|Column2|Column3|Column4|

+-------+-------+-------+-------+

|    [A]|    [B]|    [C]|    [d]|

|     |     |     |     |

+-------+-------+-------+-------+

New DF:

+-------+-------+-------+-------+

|Column1|Column2|Column3|Column4|

+-------+-------+-------+-------+

|    [A]|    [B]|    [C]|    [d]|

+-------+-------+-------+-------+

You could also create a function that will map across all the columns and do it.

Another approach (in addition to accepted answer) would be using Datasets.

For example, by having a case class:

case class MyClass(col1: Seq[String],

                   col2: Seq[Double],

                   col3: Seq[Double],

                   col4: Seq[String]) { 

    def isEmpty: Boolean = ...

}

You can represent your source as a typed structure:

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

So you could do filtering like following:

source.filter(element => !element.isEmpty) // calling class's instance method