Как рекурсивно агрегировать древовидные (иерархические) данные с помощью Spark?

Question

Учитывая набор данных, представляющий древовидную иерархическую структуру, например:

+-------+--------+
|childId|parentId|
+-------+--------+
|      1|       0|
|      2|       1|
|      3|       1|
|      4|       2|
|      5|       2|
|      6|       2|
|      7|       3|
|      8|       3|
|      9|       3|
|     10|       4|
+-------+--------+

Как его можно агрегировать, используя Spark?Таким образом, для каждого узла Дерева все его дочерние элементы, внуки и т. Д. (До листьев) могут быть агрегированы:

+--------+--------------------+-----+
|parentId|            children|count|
+--------+--------------------+-----+
|       1|[15, 9, 16, 2, 17...|   16|
|       3|[15, 9, 16, 17, 7...|    7|
|       4|    [12, 13, 10, 11]|    4|
|       7|    [15, 16, 17, 14]|    4|
|       2|[12, 13, 5, 6, 10...|    7|
|       0|[15, 9, 1, 16, 2,...|   17|
+--------+--------------------+-----+

Пример файла данных можно найти здесь .

Answer 1

Дано:

  case class Edge(childId: Int, parentId: Int)

  val edges: Dataset[Edge] = spark.read
    .option("header", value = true)
    .option("inferSchema", value = true)
    .csv("data/tree/edges.csv")
    .as[Edge]

Реализовать рекурсивный алгоритм, такой как BFS, следующим образом:

def bfs(edges: Dataset[Edge]): Dataset[Edge] = {
    @tailrec
    def helper(n: Dataset[Edge], accum: Dataset[Edge]): Dataset[Edge] = {
      val newN = n.as("n")
        .join(edges.as("plus1"), $"n.childId" === $"plus1.parentId")
        .select($"plus1.childId", $"n.parentId")
        .as[Edge]

      if (newN.count() == 0) accum else helper(newN, accum.union(newN))
    }

    edges.cache()

    helper(edges, edges)
  }

Затем вызвать следующим образом:

bfs(edges)
    .groupBy($"parentId")
    .agg(
      collect_set($"childId").alias("children"),
      countDistinct($"childId").alias("count")
    )

Полная реализация Scala здесь .Не уверен, что есть другие более простые и изящные способы сделать это.