组合 Spark Dataframe 中链接在一起的行

我有一个数据框，其行通过各种合并相互连接。 到目前为止，我已经将 DF 转换为下面的格式，我在其中执行了 groupBy“Merge_To”并将它们收集到一个数组中，然后将其连接回我的原始 DF。看起来像这样：

df1
+---+--------+---------+
|Ref|Merge_To|   Merges|
+---+--------+---------+
|  1|      \N|[3, 2, 3]|
|  2|       1|[5, 4, 6]|
|  5|       2|   [8, 7]|
| 10|      \N|   [9, 9]|
| 12|      \N|     [13]|
| 14|      \N|     [15]|
| 16|      18|     [17]|
| 17|      16|     [19]|
| 18|      \N|     [16]|
| 19|      17|     [20]|
+---+--------+---------+ 

对于参考文献 1、2、5 和 18、16、17、19、20，它们通过一条链合并在一起。这不是通过我之前完成的 groupBy 捕获的。 最终我希望我的 DF 看起来像这样，它解释了合并链：

+---+--------+------------------------+
|Ref|Merge_To|                  Merges|
+---+--------+------------------------+
|  1|      \N|[3, 2, 3, 5, 4, 6, 8, 7]|
| 10|      \N|                  [9, 9]|
| 12|      \N|                    [13]|
| 14|      \N|                    [15]|
| 18|      \N|        [16, 17, 19, 20]|
+---+--------+------------------------+

当“Merge_To”不是 \N 时，我尝试将 df1 连接到自身过滤

val arrayCombineUDF = udf((a:Seq[String], b:Seq[String]) => a ++ b )

val df1Filter = df1.filter($"Merge_To" !== "\\N").
select("Merge_To", "Merges").withColumnRenamed("Merge_To", "Chain_Ref").
withColumnRenamed("Merges", "Chain_Merges")

val df2 = df1.join(df1Filter, $"Ref" === $"Chain_Ref", "left").
withColumn("Merges", when($"Chain_Merges".isNotNull, arrayCombineUDF($"Merges", $"Chain_Merges")).
otherwise($"Merges")).
select("Ref", "Merge_To", "Merges")

df2
+---+--------+----------+------------------+
|Ref|Merge_To|Merge_From|            Merges|
+---+--------+----------+------------------+
|  1|      \N|         3|[3, 2, 3, 5, 4, 6]|
|  2|       1|        \N|   [5, 4, 6, 8, 7]|
|  5|       2|        \N|            [8, 7]|
| 10|      \N|         9|            [9, 9]|
| 12|      \N|        13|              [13]|
| 14|      \N|        \N|              [15]|
| 16|      18|        \N|          [17, 19]|
| 17|      16|        \N|          [19, 20]|
| 18|      \N|        \N|          [16, 17]|
| 19|      17|        \N|              [20]|
+---+--------+----------+------------------+

这种给出了我正在寻找的结果，但只真正说明了合并链的一层。

我还尝试将与上面相同的连接过程放入 while 循环中，试图让它重复连接。 我还尝试将 UDF 与 If 语句一起使用，希望我可以将每一行整理为合并类型，并使用它组合作为链的行。

注意：我知道数组并不不同，但我不介意，并且可以在最后对其进行排序。

EDIT这是原来的DF

+---+--------+----------+
|Ref|Merge_To|Merge_From|
+---+--------+----------+
|  1|      \N|         3|
|  2|       1|        \N|
|  3|       1|        \N|
|  4|       2|        \N|
|  5|       2|        \N|
|  6|       2|        \N|
|  7|       5|        \N|
|  8|       5|        \N|
|  9|      10|        \N|
| 10|      \N|         9|
| 11|      \N|        \N|
| 12|      \N|        13|
| 13|      \N|        \N|
| 14|      \N|        \N|
| 15|      14|        \N|
| 16|      18|        \N|
| 17|      16|        \N|
| 18|      \N|        \N|
| 19|      17|        \N|
| 20|      19|        \N|
+---+--------+----------+

11 号条目似乎擅离职守了。反正。

查看您的基础数据，这是一个分层查询，可以 在传统的 RDBMS 中解决，具有良好的功能，例如连接方式 大多数 RDBMS 中的子句或递归 WITH 视图。

您所做的这次尝试在级别 1 停止，这是问题的症结所在 问题。另外这个问题也不好解决 Spark 的并行化分区方法。分区什么 在？任何分区都可能包含与您正在查找的集合相关的数据 为了。

1. 更好的建议是在那里进行处理并 sqooping 到 Hive 表中，或者使用 JDBC 通过 Spark 进行读取。

2. 您可以按照这种记录不足的方法进行模拟https://sqlandhadoop.com/how-to-implement-recursive-queries-in-spark/ https://sqlandhadoop.com/how-to-implement-recursive-queries-in-spark/这里，这就是我有时用于 BI 后台处理的方法。

3. 如果您必须在 Spark 领域中执行此操作，那么如何使用 graphFrames 方法，但运行速度相当慢，如下所示，使用数据子集并稍微改变您的方法 - 看看您的想法：

导入 org.apache.spark.rdd.RDD

 import org.apache.spark.sql._
 import org.apache.spark.sql.functions._
 import org.graphframes._  
 sc.setCheckpointDir("/checkpoints")

// Subset of your data
val rdd = sc.parallelize( Array(("A", 1, None), ("B", 2, Some(1)), ("C", 3, Some(1)), ("D", 4, Some(2)), ("E", 5, Some(2)), ("F", 6, Some(2)),
                                ("G", 7, Some(5)), ("H", 8, Some(5)), ("X", 9, Some(10)), ("Y", 10, None), ("X2", 12, Some(13)), ("Y3", 13, None)  
                               ))
val df = rdd.toDF("v", "c", "p")

val dfV = df.select($"c".as("id"))
val dfE = df
  .withColumnRenamed("c", "src")
  .withColumnRenamed("p", "dst")

val nGraph = GraphFrame(dfV, dfE)
dfE.cache()
dfV.cache()
val res = nGraph.connectedComponents.run()
val res2 = res.join(df, res("id") === df("c"), "inner")
val res3 = res2.filter("p is not null").groupBy("component").agg(collect_list("id") as "group")
val res4 = res3.join(res2, res3("component") === res2("component") && res2("p").isNull , "inner")
res4.select($"id", $"group").show(false)

returns:

+---+---------------------+
|id |group                |
+---+---------------------+
|1  |[2, 3, 4, 5, 6, 7, 8]|
|10 |[9]                  |
|13 |[12]                 |
+---+---------------------+

我的建议：在 RDBMS 中这样做。通过这种方法使用 graphFrames 需要很长时间。