Spark sql how to explode without losing null values

by

Spark 2.2+

You can use explode_outer function:

import org.apache.spark.sql.functions.explode_outer

df.withColumn("likes", explode_outer($"likes")).show

// +---+----+--------+
// | id|name|   likes|
// +---+----+--------+
// |  1|Luke|baseball|
// |  1|Luke|  soccer|
// |  2|Lucy|    null|
// +---+----+--------+

Spark <= 2.1

In Scala but Java equivalent should be almost identical (to import individual functions use import static).

import org.apache.spark.sql.functions.{array, col, explode, lit, when}

val df = Seq(
  (1, "Luke", Some(Array("baseball", "soccer"))),
  (2, "Lucy", None)
).toDF("id", "name", "likes")

df.withColumn("likes", explode(
  when(col("likes").isNotNull, col("likes"))
    // If null explode an array<string> with a single null
    .otherwise(array(lit(null).cast("string")))))

The idea here is basically to replace NULL with an array(NULL) of a desired type. For complex type (a.k.a structs) you have to provide full schema:

val dfStruct = Seq((1L, Some(Array((1, "a")))), (2L, None)).toDF("x", "y")

val st =  StructType(Seq(
  StructField("_1", IntegerType, false), StructField("_2", StringType, true)
))

dfStruct.withColumn("y", explode(
  when(col("y").isNotNull, col("y"))
    .otherwise(array(lit(null).cast(st)))))

or 

dfStruct.withColumn("y", explode(
  when(col("y").isNotNull, col("y"))
    .otherwise(array(lit(null).cast("struct<_1:int,_2:string>")))))

Note:

If array Column has been created with containsNull set to false you should change this first (tested with Spark 2.1):

df.withColumn("array_column", $"array_column".cast(ArrayType(SomeType, true)))

More Related Contents:

  1. How to handle this scenario in spark? [closed]
  2. How to do aggregate functions, may columns and extract back
  3. How to flatten a struct in a Spark dataframe?
  4. How to create SparkSession with Hive support (fails with “Hive classes are not found”)?
  5. Spark Strutured Streaming automatically converts timestamp to local time
  6. spark broadcast variable Map giving null value
  7. How to set timezone to UTC in Apache Spark?
  8. Change output filename prefix for DataFrame.write()
  9. How do I call a UDF on a Spark DataFrame using JAVA?
  10. Retain keys with null values while writing JSON in spark
  11. How to use Column.isin in Java?
  12. Why does SparkSession execute twice for one action?
  13. Avoiding NullPointerException in Java
  14. IllegalArgumentException or NullPointerException for a null parameter? [closed]
  15. Add JAR files to a Spark job – spark-submit
  16. Java null check why use == instead of .equals()
  17. Spark: How to map Python with Scala or Java User Defined Functions?
  18. Check chains of “get” calls for null
  19. Is null check needed before calling instanceof?
  20. Spark Error – Unsupported class file major version
  21. Apache Spark – foreach Vs foreachPartition When to use What?
  22. Why use Optional.of over Optional.ofNullable?
  23. PySpark: java.lang.OutofMemoryError: Java heap space
  24. TaskSchedulerImpl: Initial job has not accepted any resources;
  25. Is null an Object?
  26. Hibernate order by with nulls last
  27. How can I force Spark to execute code?
  28. Reading data from Azure Blob with Spark
  29. Difference between SparkContext, JavaSparkContext, SQLContext, and SparkSession?
  30. Spark: get number of cluster cores programmatically

Leave a Comment