在C++中高效地对一组集合进行交集操作

Question

我有一组std :: set集合，我想以最快的方式找到该集合中所有集合的交集。该集合中的集合数量通常非常小（〜5-10），每个集合中的元素数量通常少于1000，但有时可能会增加到约10000。但是，我需要执行这些交集数万次，尽可能快地完成。 我尝试了以下几种方法进行基准测试：

1. 在std :: set对象中进行就地交集，该对象最初复制第一个集合。然后对于后续集合，它会遍历自身和集合的第i个元素，并根据需要从自身中删除项目。
2. 使用std :: set_intersection进入临时std :: set，将内容交换到当前集合，然后再次查找当前集合与下一个集合的交集并插入临时集，依此类推。
3. 像1）中一样手动迭代所有集合的所有元素，但是使用vector作为目标容器而不是std :: set。
4. 与4相同，但使用std :: list而不是vector，怀疑list将提供更快的从中间删除。
5. 使用哈希集（std :: unordered_set）并检查所有集合中的所有项。

结果表明，在每个集合中的元素数量较小时，使用vector略微更快; 在集合较大时，list略微更快。就地使用集比两者都慢得多，其次是set_intersection和哈希集。是否有更快的算法/数据结构/技巧来实现此目的？如果需要，我可以发布代码片段。 谢谢!

- Paresh

---

问题实际上取决于您是否需要找到许多共同元素，因为这会改变人们可以想出的“最佳”结构。例如，第六种方法可以简单地使用std::unordered_map并计算每个元素的出现次数。它在元素总数为O（N）的情况下运行。然后，您只需选择具有与集合数量相等的总数的元素，这在不同元素的数量为O（M）的情况下进行。不知道它的表现如何。 - Matthieu M.

@MatthieuM。这种方法将以未排序的方式给出结果集。幸运的是，我有两个用例，一个需要按排序顺序返回结果，另一个则不需要。如果这种方法足够快，我至少可以在不需要对交集进行排序的情况下使用它。 - Paresh

你可以尝试这个想法。最坏情况下是线性的（如果集合主要包含相同元素，则无法避免），但如果交集很小，速度会更快。 - Daniel Fischer

@DanielFischer 谢谢！由于Dietmar在下面的回答中，我也考虑过在数组搜索时使用二分查找。但最坏情况下的减速是一个担忧。您提出了一个非常好的启发式/估计方法，使这成为一种混合方法。实际上，由于少量额外的计算，这只比向量方法（第3部分）略慢，但如果后续集合的大小足够大，则明显是所有方法中最快的！非常好的想法！ - Paresh

@DanielFischer 如果这是一个答案，我会接受的。 - Paresh

Answer 1

您可以尝试一个std::set_intersection()的推广版：使用迭代器来处理所有集合：

1. 如果任何一个迭代器已经到达了其对应集合的end()，那么搜索结束。因此，可以假设所有迭代器都是有效的。
2. 以第一个迭代器的值作为下一个候选值x。
3. 移动迭代器列表，并std::find_if()第一个大于或等于x的元素。
4. 如果这个值比x更大，则将其作为新的候选值并在迭代器序列中再次搜索。
5. 如果所有迭代器都在值x上，那么就找到了交集的一个元素：记录它，增加所有迭代器的值，重新开始搜索。

- Dietmar Kühl

---

@MatthieuM。在这种情况下不是这样的，find_if平均而言永远不需要超过两个元素，并且因此是_O_(1)，而???er_bound是_O_(log n)。 - leftaroundabout

@leftaroundabout 谢谢！我不明白为什么 find_if 平均来说永远不需要超过两个元素的推进？ - Paresh

@leftaroundabout：像Paresh一样，我想知道这2个元素是从哪里来的（我可能错过了一些显而易见的东西）。在我看来，这似乎取决于数据的分布，不是吗？例如，假设我有100个元素的集合和另一个涵盖相同范围的1000个元素。然后平均每步需要跳过大集合中约10个元素。 - Matthieu M.

@MatthieuM.，Paresh：我认为它对于某些类别的分布在每个集合上平均工作也应该是可行的，但我不确定是否能够证明。 - leftaroundabout

接受这个方案，因为它有进一步优化的空间，并且如果某些用例只需要交集的前几个元素，则可以用于早期终止。 - Paresh

Answer 2

夜晚是一个好的顾问，我想我可能有一个主意 ;)

• 现在内存比CPU慢得多，如果所有数据都适合L1缓存，那就没什么大问题了，但很容易溢出到L2或L3：5个1000元素的集合已经是5000个元素，意味着5000个节点，而一个集合节点至少包含3个指针+对象（即，在32位机器上至少16字节，在64位机器上为32字节）=>这至少需要80k内存，而最近的CPU只有32k用于L1D，所以我们已经溢出到L2了
• 前面的事实加剧了这个问题：集合节点可能分散在内存中，而不是紧密地打包在一起，这意味着缓存行的一部分被填充了完全无关的内容。这可以通过提供一个保持节点彼此靠近的分配器来缓解。
• 这进一步加剧了CPU更擅长顺序读取（它们可以在你需要之前预取内存，因此你不必等待它）而不是随机读取（而树结构不幸地导致相当随机的读取）

这就是为什么速度很重要时，vector（或者可能是deque）是如此伟大的结构：它们与内存非常相似。因此，我肯定会推荐使用vector作为我们的中间结构；尽管需要注意只从一个极端插入/删除以避免重定位。

所以我想到了一个相当简单的方法：

#include <cassert>

#include <algorithm>
#include <set>
#include <vector>

// Do not call this method if you have a single set...
// And the pointers better not be null either!
std::vector<int> intersect(std::vector< std::set<int> const* > const& sets) {
    for (auto s: sets) { assert(s && "I said no null pointer"); }

    std::vector<int> result; // only return this one, for NRVO to kick in

    // 0. Check obvious cases
    if (sets.empty()) { return result; }

    if (sets.size() == 1) {
        result.assign(sets.front()->begin(), sets.front()->end());
        return result;
    }


    // 1. Merge first two sets in the result
    std::set_intersection(sets[0]->begin(), sets[0]->end(),
                          sets[1]->begin(), sets[1]->end(),
                          std::back_inserter(result));

    if (sets.size() == 2) { return result; }


    // 2. Merge consecutive sets with result into buffer, then swap them around
    //    so that the "result" is always in result at the end of the loop.

    std::vector<int> buffer; // outside the loop so that we reuse its memory

    for (size_t i = 2; i < sets.size(); ++i) {
        buffer.clear();

        std::set_intersection(result.begin(), result.end(),
                              sets[i]->begin(), sets[i]->end(),
                              std::back_inserter(buffer));

        swap(result, buffer);
    }

    return result;
}

看起来正确，但我无法保证其速度，显然。

- Matthieu M.

---

谢谢！内存的紧凑性是我尝试原问题中选项3的原因：使用vector作为中间容器，就像您所做的一样。不同之处在于，您使用了set_intersection，它需要两个vectors，而我只保留了一个vector，缺点是我必须从中间删除。尽管您的方法理论上应该更快，但我猜测像连续内存、缓存（1个数组与2个数组）等复杂因素使得这比我尝试过的选项3和4要慢。当然，里程可能会因数据而异。 - Paresh

赞同你考虑内存和缓存，以及给出清晰的解释！顺便说一下，我正在考虑使用向量来代替std::set，并按排序顺序插入到向量中，如果比较起来是可行的话。紧凑性可能会使它变得相当快，而交集肯定会更快。 - Paresh