基于 Scala 中的键列合并 Spark 数据帧行

Question

我有一个包含 2 列的流式数据框。一个表示为字符串的键列和一个对象列，它是一个包含一个对象元素的数组。我希望能够使用相同的键合并 Dataframe 中的记录或行，以便合并的记录形成一个对象数组。

Dataframe

----------------------------------------------------------------
|key    | objects                                              |
----------------------------------------------------------------
|abc    | [{"name": "file", "type": "sample", "code": "123"}]  |
|abc    | [{"name": "image", "type": "sample", "code": "456"}] |
|xyz    | [{"name": "doc", "type": "sample", "code": "707"}]   |
----------------------------------------------------------------


Merged Dataframe

-------------------------------------------------------------------------
|key   |  objects                                                        |
-------------------------------------------------------------------------
|abc   |    [{"name": "file", "type": "sample", "code": "123"}, {"name": 
            "image", "type": "sample", "code": "456"}]                   |
|xyz   |   [{"name": "doc", "type": "sample", "code": "707"}]            |
--------------------------------------------------------------------------

执行此操作的一个选项是将其转换为 PairedRDD 并应用 reduceByKey 函数，但如果可能，我更喜欢使用 Dataframes 执行此操作，因为它会更优化。有什么方法可以在不影响性能的情况下使用 Dataframes 做到这一点？

Answer 1

假设列 objects 是单个 JSON 字符串的数组，以下是如何将 objects 与 key 合并：

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

case class Obj(name: String, `type`: String, code: String)

val df = Seq(
    ("abc", Obj("file", "sample", "123")),
    ("abc", Obj("image", "sample", "456")),
    ("xyz", Obj("doc", "sample", "707"))
  ).
  toDF("key", "object").
  select($"key", array(to_json($"object")).as("objects"))

df.show(false)
// +---+-----------------------------------------------+
// |key|objects                                        |
// +---+-----------------------------------------------+
// |abc|[{"name":"file","type":"sample","code":"123"}] |
// |abc|[{"name":"image","type":"sample","code":"456"}]|
// |xyz|[{"name":"doc","type":"sample","code":"707"}]  |
// +---+-----------------------------------------------+

df.groupBy($"key").agg(collect_list($"objects"(0)).as("objects")).
  show(false)
// +---+---------------------------------------------------------------------------------------------+
// |key|objects                                                                                      |
// +---+---------------------------------------------------------------------------------------------+
// |xyz|[{"name":"doc","type":"sample","code":"707"}]                                                |
// |abc|[{"name":"file","type":"sample","code":"123"}, {"name":"image","type":"sample","code":"456"}]|
// +---+---------------------------------------------------------------------------------------------+