Spark - удалить дубликаты внутри поля массива json

Question

Я делюсь кодом, который у меня есть:

// define a case class
case class Zone(id: Int, team: String, members: Int ,name: String,  lastname: String)
        val df = Seq (
         (1,"team1", 3, "Jonh", "Doe"),
         (1,"team2", 4, "Jonh", "Doe"),
         (1,"team3", 5, "David", "Luis"),
         (2,"team4", 6, "Michael", "Larson"))
         .toDF("id", "team", "members", "name",  "lastname").as[Zone]

val df_grouped = df
   .withColumn("team_info", to_json(struct(col("team"), col("members"))))
   .withColumn("users", to_json(struct(col("name"), col("lastname"))))
   .groupBy("id")
   .agg(collect_list($"team_info").alias("team_info"), collect_list($"users").alias("users"))

df_grouped.show    
    +---+--------------------+--------------------+
    | id|           team_info|               users|
    +---+--------------------+--------------------+
    |  1|[{"team":"team1",...|[{"name":"Jonh","...|
    |  2|[{"team":"team4",...|[{"name":"Michael...|
    +---+--------------------+--------------------+

Мне нужно удалить дубликаты внутри столбца "пользователи", потому что в моем случае, если json внутри массива точно такие же, являются дубликатами. Есть ли способ сделать это, изменив значение этого столбца с помощью df.withColumn или любым другим подходом?

Answer 1

Это, вероятно, не самое элегантное решение, но оно должно работать:

import org.apache.spark.sql.types._
import org.apache.spark.sql.Encoders

val df = sc.parallelize(
  Array("[{\"name\":\"John\",\"lastName\":\"Doe\"},{\"name\":\"John\",\"lastName\":\"Doe\"},{\"name\":\"David\",\"lastName\":\"Luis\"}]")
).toDF("users")

case class Users(name: String, lastName: String)
val schema = ArrayType(Encoders.product[Users].schema)

df.withColumn("u", from_json($"users", schema))
  .select("u")
  .as[Array[Users]]
  .map(_.distinct)
  .toDF("u")
  .withColumn("users", to_json($"u"))
  .select("users")

Предполагая, что у ваших пользователей будет больше атрибутов, чем в вашем примере, просто добавьте эти атрибуты в класс дел. Пока типы просты, Encoder должен выводить схему автоматически.

Answer 2

Вы можете использовать встроенные функции explode и dropDuplicates

Искры DropDuplicates на основе поля массива JSON