Я пытаюсь решить проблему, так что у меня есть такой набор данных:
(1, 3)
(1, 4)
(1, 7)
(1, 2) <-
(2, 7) <-
(6, 6)
(3, 7) <-
(7, 4) <-
...
Так как (1 -> 2) и (2 -> 7) хотелось бы заменить набор (2, 7) на (1, 7) аналогично, (3 -> 7) и (7 -> 4) тоже заменить (7,4) на (3, 4)
Следовательно, мой набор данных становится
(1, 3)
(1, 4)
(1, 7)
(1, 2)
(1, 7)
(6, 6)
(3, 7)
(3, 4)
...
Любая идея, как решить или решить эту проблему?
Спасибо
Эта задача выглядит как транзитивное замыкание графа, представленного в виде распределенного списка ребер.
Одной из ключевых особенностей Spark по сравнению со старой версией Hadoop MR является то, что Spark поддерживает интерактивные алгоритмы. Чтобы решить подобную задачу обхода графа, мы используем эту возможность в рекурсивной функции:
def closure(rdd:RDD[(Int, Int)]):RDD[(Int,Int)] = {
val transitiveValues = rdd.map(_.swap).join(rdd).filter{case (_,(x,y)) => x != y}
if (transitiveValues.isEmpty) {
rdd
} else {
val usedTransitions = transitiveValues.flatMap{case (a,(x,y)) => Seq((x,a),(a,y))}
val newTransitions = transitiveValues.map{case (a,(x,y)) => (x,y)}
closure(rdd.subtract(usedTransitions).union(newTransitions)).distinct
}
}
Это не совсем приводит к ожидаемому выше результату, потому что нет понятия приоритета (неявного порядка), поэтому closure((1, 2),(2, 7)) = (1,7), а не (1, 2), (1, 7), как ожидалось выше. Заказ может быть добавлен за счет дополнительной сложности. Также он не поддерживает циклические графики (с циклами).
Этот алгоритм должен служить только отправной точкой для настройки на конкретные внутренние требования.
