Преобразование нулевых значений в пустой массив в Spark DataFrame

У меня есть пакет данных Spark, где один столбец представляет собой массив целых чисел. Столбец является нулевым, потому что он исходит из левого внешнего соединения. Я хочу преобразовать все пустые значения в пустой массив, поэтому мне больше не придется иметь дело с нулями.

Я думал, что смогу сделать это так:

val myCol = df("myCol")
df.withColumn( "myCol", when(myCol.isNull, Array[Int]()).otherwise(myCol) )

Однако это приводит к следующему исключению:

java.lang.RuntimeException: Unsupported literal type class [I [[email protected]
at org.apache.spark.sql.catalyst.expressions.Literal$.apply(literals.scala:49)
at org.apache.spark.sql.functions$.lit(functions.scala:89)
at org.apache.spark.sql.functions$.when(functions.scala:778)

По-видимому, типы массивов не поддерживаются функцией when. Есть ли другой простой способ преобразования нулевых значений?

В случае, если это имеет значение, вот схема для этого столбца:

|-- myCol: array (nullable = true)
|    |-- element: integer (containsNull = false)

Ответы

Ответ 1

Вы можете использовать UDF:

import org.apache.spark.sql.functions.udf

val array_ = udf(() => Array.empty[Int])

в сочетании с WHEN или COALESCE:

df.withColumn("myCol", when(myCol.isNull, array_()).otherwise(myCol))
df.withColumn("myCol", coalesce(myCol, array_())).show

В последних версиях вы можете использовать функцию array:

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

df.withColumn("foo", array().cast("array<integer>"))

Обратите внимание, что это будет работать только в том случае, если разрешено преобразование string в нужный тип.

Ответ 2

С небольшой модификацией подхода zero323 я смог сделать это без использования udf в Spark 2.3.1.

val df = Seq("a" -> Array(1,2,3), "b" -> null, "c" -> Array(7,8,9)).toDF("id","numbers")
df.show
+---+---------+
| id|  numbers|
+---+---------+
|  a|[1, 2, 3]|
|  b|     null|
|  c|[7, 8, 9]|
+---+---------+

val df2 = df.withColumn("numbers", coalesce($"numbers", array()))
df2.show
+---+---------+
| id|  numbers|
+---+---------+
|  a|[1, 2, 3]|
|  b|       []|
|  c|[7, 8, 9]|
+---+---------+

Ответ 3

Альтернатива без UDF для использования, когда тип данных, в котором вы хотите, чтобы элементы массива не могли быть StringType из StringType является следующим:

import pyspark.sql.types as T
import pyspark.sql.functions as F

df.withColumn(
    "myCol",
    F.coalesce(
        F.col("myCol"),
        F.from_json(F.lit("[]"), T.ArrayType(T.IntegerType()))
    )
)

Вы можете заменить IntegerType() любым типом данных, также сложным.