你完全找对方向了！把df_products先扁平化转换成你描述的三列结构（product_PK, rec_product_PK, rank），正是解决这个性能瓶颈的核心方案。原来的代码之所以在数组元素较多时陷入停滞，主要是因为explode后直接做关联，加上嵌套字段的匹配逻辑，让Spark难以高效处理，最终导致中间数据膨胀、Shuffle开销剧增。

第一步：高效扁平化df_products #

先把嵌套的products数组展开，直接提取出每个数组对里的推荐产品PK和排名，转换成扁平化的结构——这一步是优化的关键：

val flattenedProducts = df_products
  // 先展开数组，得到每个单独的产品-排名对
  .select($"product_PK", explode($"products").as("product_pair"))
  // 从展开的结构体中提取顶级字段，避免后续关联时的嵌套解析开销
  .select(
    $"product_PK",
    $"product_pair._1".as("rec_product_PK"), // 对应数组对的第一个元素：推荐产品PK
    $"product_pair._2".as("rank") // 对应数组对的第二个元素：排名值
  )

// 缓存扁平化后的数据集，避免后续重复计算数组展开的逻辑
flattenedProducts.cache()

这样处理后，flattenedProducts就完全是你想要的结构了：

+----------+-----------------+------+
|product_PK|    rec_product_PK|  rank|
+----------+-----------------+------+
|       111|              222|    66|
|       111|              333|    55|
|       222|              333|    24|
|       222|              444|    77|
...
+----------+-----------------+------+

第二步：优化后的关联操作 #

现在用扁平化后的数据集和df做关联，因为两个表的关联字段都是顶级字段，Spark的优化器可以选择更高效的关联策略（比如Broadcast Hash Join或Sort Merge Join），性能会比原来的嵌套字段关联提升很多：

val result = df.as("df2")
  .join(
    flattenedProducts.as("df1"),
    // 用顶级字段做关联条件，避免嵌套解析开销
    $"df2.product_PK" === $"df1.product_PK" && $"df2.rec_product_PK" === $"df1.rec_product_PK",
    "left"
  )
  .select(
    $"df2.product_PK",
    $"df2.rec_product_PK",
    // 处理左关联时的空值，用0.0填充
    coalesce($"df1.rank", lit(0.0)).as("rank_product")
  )

为什么原来的代码性能差？ #

• 嵌套字段的解析开销：原来的关联条件中用到$"df1.products.product_PK"，每次匹配都要解析嵌套的结构体，比直接访问顶级字段的开销大得多。
• 中间数据膨胀：explode后直接做关联，Spark无法提前过滤无效数据，导致中间数据集急剧增大，大幅增加了Shuffle阶段的IO和计算开销。而先扁平化并缓存，相当于提前完成了数组展开的计算，后续关联可以直接复用预处理后的结果。

额外的性能调优建议 #

• 缓存策略调整：如果flattenedProducts的大小适合内存存储，用cache()即可；如果内存不足，改用persist(StorageLevel.MEMORY_AND_DISK)，避免内存溢出。
• 集群参数调优：针对800MB的数据集，可以适当调整spark.sql.shuffle.partitions（默认200，建议设为与集群CPU核心数匹配的值，比如64或128），减少Shuffle产生的小文件；同时调大spark.driver.memory和spark.executor.memory，避免内存不足导致的磁盘溢出。
• 避免冗余操作：原来的代码中将df_products转为临时视图再查询，多了一层不必要的转换，直接使用DataFrame API操作更高效。

内容的提问来源于stack exchange，提问作者Markus