高效的分布式算法来合并具有共同元素的集合答案

【问题标题】：Efficient distributed algorithm to merge sets with common elements高效的分布式算法来合并具有共同元素的集合
【发布时间】：2016-05-27 14:55:40
【问题描述】：

我正在 Flink 上进行 MinHash LSH 的分布式实现，作为最后一步，我需要合并一些集群，这些集群被标识为它们之间相似的元素集。

所以我有一个分布式集合作为输入，我需要一种算法来有效地将集合与公共元素合并。给定 Flink 的计算模型，算法可能是迭代的，不一定是 map-reduce 之类的。

这里是一个例子：

来自{{1{1,2}},{2,{2,3}},{3,{4,5},{4{1,27}}}} 的结果应该是{1,2,3,27},{4,5}，因为集合#1、#2 和#4 至少有一个共同的元素。

【问题讨论】：

所以这是传递的？ IE。如果 A 和 B 有一个共同元素，而 B 和 C 有一个不同的共同元素，你想要 A union B union C 在结果中吗？
我们表现得好像它是传递的，是的

标签： algorithm scala merge distributed-computing apache-flink

【解决方案1】：

这里有一个想法：作为 Flink 一部分的 Gelly 有一个连接组件查找器。制作一个图，其中包含每个集合元素的节点和以最简单的方式连接每个集合的元素的边，例如对于 {a, b, c, d, ...} 添加 [a,b], [a,c], [a,d], [a,... 。现在找到连接的组件。他们的节点给出了你正在寻找的集合。

编辑如果您担心从集合转换为图表并返回的性能影响（尽管这种担心是过早的优化；您应该尝试一下），重新实现 Gelly 的令牌推送方案就足够简单了。这是如何工作的。您的示例中已经有了标记：集合编号。让 S[i] 设置为 i 在您的示例中显示。例如。 S[1] = {1,2}。令 R 是一个逆多重映射，它将每个集合元素带到它所属的集合集合中。例如。 R[2] = {1,2} 在您的示例中。令 T[i] 为集合 i 可通过传递非空交集“链接”到达的元素。然后计算：

T[i] = S[i] for all i // with no links at all, a set reaches its own elements
loop
  for all i, Tnew[i] = \union_{ x \in T[i] } S[R[x]]  // add new reachables
  exit if Tnew == T
  T = Tnew
end loop

完成后，地图 T 的不同值就是您想要的答案。最大迭代次数应该是 log |U|其中 U 是集合元素的全域。

【讨论】：

我考虑过，它可能会起作用，但 Gelly 尚未成熟，但我相信总体而言转换成本太高。无论如何我都会尝试，因为它可能表现得足够好。
@Chobeat 好吧，如果您正在寻找分布式算法，那么像 Gelly 所做的事情可能是您能做的最好的事情。不成熟总比从头开始好。 nb：我改变了连接图中集合元素的方式，以从 Gelly 的算法中获得更好的性能。
@Chobeat 好的，如果您想尝试一下，我将 Gelly 算法转换回您的问题。
明天我将花一天时间尝试这里提出的建议，从您的开始。非常感谢。 :)
好的，我试过 Gelly，我想我会按照你的建议来避免早期优化。谢谢。

【解决方案2】：

只是一个想法，可能有更好的方法，但是这个怎么样：

在映射步骤中，您为每个集合中的每个元素发出一个键值对，例如：element -> other elements
在减少步骤中，您收集那些其他元素并丢弃重复项
重复直到数据结构停止变化

第一次迭代后，您的数据将如下所示：

1 -> 2, 27
2 -> 1,3
3 -> 2
4 -> 5
5 -> 4
27 -> 1

第二个之后：

1 -> 2, 3, 27
2 -> 1, 3, 27
3 -> 1, 2
4 -> 5
5 -> 4
27 -> 1, 2

最后在第三个之后：

1 -> 2, 3, 27
2 -> 1, 3, 27
3 -> 1, 2, 27
4 -> 5
5 -> 4
27 -> 1, 2, 3

我目前没有办法确定更改何时停止。

要仅获取每个结果的一个副本，您可以删除所有键大于任何其他元素的位置。

【讨论】：

【解决方案3】：

如果您有N 集合，每个集合大约有M 元素，那么如果重复很少见，那么简单的方法（测试每个集合的每个元素）是O(N^2 * M^2)。但是，如果您实际上只有 R << N*M 不同的元素，那么它并没有那么糟糕：一旦发现某些东西，您就可以停止测试，这种情况发生在 N*M 比较之后，而只有 R，所以您只能“仅” O(N*N*R)。但是，如果集合实际上只存在于L 组中，则您不必将每个集合与其他集合进行测试，因为一旦找到正确的组，您就会停止。所以它更像O(N*L*R) + O(N*M)（第二个术语实际上是在找到要添加的正确组后将元素添加到组中）。

如果你将每个元素映射到它包含的集合列表——你可以在O(N*M) time 完成——那么你可以遍历每个元素的集合树，最多访问每个不同的元素大约一次（即其中R），并且每个人访问提到它的每个集合（结果大约是N*M/R）并添加其所有元素（但只有一次！），总体而言是O(N*M)如果您小心不要多次添加相同的集合，则需要时间。（您需要一个集合的包装器，这样您就可以判断您是否已经访问过它们。）这样会更快，但如果 L*R 非常小，您可能不会在意。

在 Scala 中，从元素到树的映射的核心类似于

case class W(s: Set[Int]) { var visited: Boolean = false }
def tree(ss: Seq[Set[Int]]) = {
  var m = new collection.mutable.HashMap[Int, List[W]]
  ss.foreach{ s =>
    s.foreach{i =>
      m(i) = W(s) :: m.getOrElse(i, Nil)
    }
  }
}

并且遍历组更复杂，但基本想法是保留您看到的元素的地图，如果您遇到其中一个，则不要继续遍历，并且还要跟踪您是否遍历了一个通过在合并元素时在 W 包装器中设置标志来设置。

【讨论】：