Как сгруппировать определенный столбец и извлечь всю строку в виде строки JSON в Scala Spark?

Question

Я пытаюсь собрать набор данных в формате JSON

val df = spark.sql("select invn_ctl_nbr,cl_id,department from pi_prd.table1 where batch_run_dt='20190101' and batchid = '20190101001' limit 10").toJSON.rdd

Сгенерированный результат в формате Array [String]:

Array({"invn_ctl_nbr":"1","cl_id":"AK","department":"Dept1"}
{"invn_ctl_nbr":"2","cl_id":"AF","department":"Dept1"}
{"invn_ctl_nbr":"3","cl_id":"AF","department":"Dept2"}
{"invn_ctl_nbr":"4","cl_id":"AK","department":"Dept3"})

Далее, я пытаюсь сгруппировать данные таким образом, чтобы они дали мне результат в следующем формате:

Map<key, List<data>>

Чтобы привести пример:

Map<AK=[{"invn_ctl_nbr":"1","cl_id":"AK","department":"Dept1"},{"invn_ctl_nbr":"4","cl_id":"AK","department":"Dept3"}],AF=[{"invn_ctl_nbr":"2","cl_id":"AF","department":"Dept1"},{"invn_ctl_nbr":"3","cl_id":"AF","department":"Dept2"}]>

Я уже пробовал решение, указанное в ссылке . Код, который я использовал для получения требуемого результата, был:

import org.json4s._
import org.json4s.jackson.Serialization.read

case class cC(invn_ctl_nbr: String,cl_id: String,department: String)

val rdd_new = df.map(m => {
implicit val formats = DefaultFormats
val parsedObj = read[cC](m)
(parsedObj.srk_clm_id, m)
})

rdd_new.collect.groupBy(_._1).map(m => (m._1,m._2.map(_._2).toList))

но это дает мне ошибку ниже:

org.json4s.package$MappingException: Parsed JSON values do not match with class constructor
args=
arg types=
constructor=public cC($iw,java.lang.String,java.lang.String,java.lang.String)

Мое сопоставление верно для того, что я указал в классе case и что идет от rdd, не уверен, что именно здесь мне не хватает. Кто-нибудь, пожалуйста, помогите в решении этой проблемы? Спасибо.

Answer 1

Вы можете использовать groupBy с struct, to_json и collect_list, чтобы получить желаемый результат

Фрейм данных df здесь эквивалентен вашему spark.sql("select query")

val df = Seq(
  ("1", "AK", "Dept1"),
  ("2", "AF", "Dept1"),
  ("3", "AF", "Dept2"),
  ("4", "AK", "Dept3")
).toDF("invn_ctl_nbr","cl_id","department")


val result = df.groupBy($"cl_id")
  .agg(to_json(collect_list(struct(df.columns.map(col(_)) : _*))))
  .rdd.map(x => (x.getString(0), x.get(1))).collectAsMap()

Вывод (результат):

Map(AF -> [{"invn_ctl_nbr":"2","cl_id":"AF","department":"Dept1"},{"invn_ctl_nbr":"3","cl_id":"AF","department":"Dept2"}], AK -> [{"invn_ctl_nbr":"1","cl_id":"AK","department":"Dept1"},{"invn_ctl_nbr":"4","cl_id":"AK","department":"Dept3"}])

Надеюсь, это поможет!