Выберите последний элемент массива в DataFrame

Question

Я работаю над проектом и имею дело с некоторыми вложенными датами JSON со сложной схемой / структурой данных.По сути, я хочу отфильтровать один из столбцов в кадре данных, чтобы я выбрал последний элемент в массиве.Я полностью застрял на том, как это сделать.Я надеюсь, что это имеет смысл.

Ниже приведен пример того, чего я пытаюсь достичь:

val singersDF = Seq(
  ("beatles", "help,hey,jude"),
  ("romeo", "eres,mia"),
  ("elvis", "this,is,an,example")
).toDF("name", "hit_songs")

val actualDF = singersDF.withColumn(
  "hit_songs",
  split(col("hit_songs"), "\\,")
)

actualDF.show(false)
actualDF.printSchema() 

+-------+-----------------------+
|name   |hit_songs              |
+-------+-----------------------+
|beatles|[help, hey, jude]      |
|romeo  |[eres, mia]            |
|elvis  |[this, is, an, example]|
+-------+-----------------------+
root
 |-- name: string (nullable = true)
 |-- hit_songs: array (nullable = true)
 |    |-- element: string (containsNull = true)

Конечной целью для вывода будет следующее, чтобы выбрать последний "строка "в массиве hit_songs.

Меня не волнует, как будет выглядеть схема впоследствии.

+-------+---------+
|name   |hit_songs|
+-------+---------+
|beatles|jude     |
|romeo  |mia      |
|elvis  |example  |
+-------+---------+

Answer 1

Вот один подход:

val actualDF = Seq(
  ("beatles", Seq("help", "hey", "jude")),
  ("romeo", Seq("eres", "mia")),
  ("elvis", Seq("this", "is", "an", "example"))
).toDF("name", "hit_songs")

import org.apache.spark.sql.functions._

actualDF.withColumn("total_songs", size($"hit_songs")).
  select($"name", $"hit_songs"($"total_songs" - 1).as("last_song"))
// +-------+---------+
// |   name|last_song|
// +-------+---------+
// |beatles|     jude|
// |  romeo|      mia|
// |  elvis|  example|
// +-------+---------+

Answer 2

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

val lastElementUDF = udf((array: Seq[String]) => array.lastOption)

actualDF.withColumn("hit_songs", lastElementUDF($"hit_songs"))

array.lastOption вернет None или Some, а array.last сгенерирует исключение, если массив пуст.

Answer 3

Вы можете использовать функцию size для вычисления индекса нужного элемента в массиве, а затем передать его в качестве аргумента Column.apply (явно или неявно):

import org.apache.spark.sql.functions._
import spark.implicits._

actualDF.withColumn("hit_songs", $"hit_songs".apply(size($"hit_songs").minus(1)))

Или:

actualDF.withColumn("hit_songs", $"hit_songs"(size($"hit_songs").minus(1)))