SPARK - 在数组中的值上连接 2 个数据帧

Posted 2023-04-17

【中文标题】SPARK - 在数组中的值上连接 2 个数据帧

【英文标题】：SPARK - Joining 2 dataframes on values in an array

【发布时间】：2019-12-30 17:05:05

【问题描述】：

我找不到一个简单而优雅的解决方案。

我在这个专栏中有一个 df1：

|-- guitars: array (nullable = true)
 |    |-- element: long (containsNull = true)

我有一个由吉他制成的 df2，以及一个与我的 df 1 中的 Long 匹配的 id。

root
 |-- guitarId: long (nullable = true)
 |-- make: string (nullable = true)
 |-- model: string (nullable = true)
 |-- type: string (nullable = true)

显然，我想加入我的两个 dfs，而不是一个 long 数组，我想要一个来自 df2 的 struct 吉他数组。

我正在使用array_contains() 加入两个 dfs，但 spark 正在爆炸结果 df 中 n 行中 df1 中的 n Long 数组。

之前

|   2|Eric Clapton| [1, 5]|               [,,,]|

之后

|   2|Eric Clapton| [1, 5]|               [,,,]|       5|Fender|Stratocaster|            Electric|
|   2|Eric Clapton| [1, 5]|               [,,,]|       1|Gibson|          SG|            Electric|

将这个 Long 数组列从其他数据帧转换为 struct 数组列的最优雅的解决方案是什么？

理想

|   2|Eric Clapton|[[Fender, Stratocaster, Electric],[Gibson, SG, Electric]]|               [,,,]|

提前致谢

（顺便说一句，第一个问题，谦虚点：P）

【问题讨论】：

数据模型是否正确？ df1 吉他包含许多 ID，其中一个/多个应该匹配？你可能有一个笛卡尔连接。

【参考方案1】：

array_contains() 有效，之后只需要按玩家分组结果即可。

让我们从两个数据集开始，一个用于演奏者，一个用于吉他：

val player = Seq(("Eric Clapton", Array(1,5)), ("Paco de Lucia", Array(1,2)), ("Jimi Hendrix", Array(3))).toDF("player", "guitars")
val guitar = Seq((1, "Gibson", "SG", "Electric"), (2, "Faustino Conde", "Media Luna", "Acoustic"), (3, "Pulsebeatguitars", "Spider", "Electric"), (4, "Yamaha", "FG800", "Acoustic"), (5, "Fender", "Stratocaster", "Electric")).toDF("guitarId", "make", "model", "type")

+-------------+-------+
|       player|guitars|
+-------------+-------+
| Eric Clapton| [1, 5]|
|Paco de Lucia| [1, 2]|
| Jimi Hendrix|    [3]|
+-------------+-------+
+--------+----------------+------------+--------+
|guitarId|            make|       model|    type|
+--------+----------------+------------+--------+
|       1|          Gibson|          SG|Electric|
|       2|  Faustino Conde|  Media Luna|Acoustic|
|       3|Pulsebeatguitars|      Spider|Electric|
|       4|          Yamaha|       FG800|Acoustic|
|       5|          Fender|Stratocaster|Electric|
+--------+----------------+------------+--------+

为了让分组操作更简单一点，想法是在join之前将吉他数据集的三列组合成一个struct：

val guitar2 = guitar.withColumn("guitar", struct('make, 'model, 'type))

加入后，我们将结果按玩家分组，得到正确的结果：

player.join(guitar2, expr("array_contains(guitars, guitarId)"))
  .groupBy("player")
  .agg(collect_list('guitar))
  .show(false)

打印

+-------------+----------------------------------------------------------------+
|player       |collect_list(guitar)                                            |
+-------------+----------------------------------------------------------------+
|Jimi Hendrix |[[Pulsebeatguitars, Spider, Electric]]                          |
|Eric Clapton |[[Gibson, SG, Electric], [Fender, Stratocaster, Electric]]      |
|Paco de Lucia|[[Gibson, SG, Electric], [Faustino Conde, Media Luna, Acoustic]]|
+-------------+----------------------------------------------------------------+

【讨论】：

谢谢。更进一步，您知道我在哪里可以找到有关这些机制的一些好的资源（即：“collect_list”、“expr()”等）吗？

我通常从浏览官方文档开始：spark.apache.org/docs/latest/api/scala/…。如果某个函数看起来有帮助，Google 或 *** 通常会提供一些示例

@werner 我尝试了上述解决方案来解决一些类似的情况。但我得到了模棱两可的错误。我的问题在这里***.com/questions/66791957/…。你能帮我解决这个问题吗？