执行向量c ++的交集答案

【问题标题】：Performing intersection of vector c++执行向量c ++的交集
【发布时间】：2015-09-12 22:35:00
【问题描述】：

我有 200 个大小从 1 到 4000000 的向量存储在 vecOfVec 中。我需要将这些向量与大小为 9000+ 个元素的单个向量“vecSearched”相交。我尝试使用以下代码做同样的事情，但是使用 perf 工具我发现我正在做的交集是我代码中的瓶颈。有什么方法可以让我进行有效的交叉路口

#include <cstdlib>
#include <iostream>
#include <vector>

using namespace std;

int main(int argc, char** argv) {
  vector<vector<unsigned> > vecOfVec; //contains 120 vectors of size ranging from 1 to 2000000 elements. All vectors in vecOfVec are sorted
  vector<unsigned> vecSearched; //vector searched in contains 9000+ elements. Vectors in vecSearched are sorted
  for(unsigned kbt=0; kbt<vecOfVec.size(); kbt++)
  {                                  
     //first find first 9 values spaced at equi-distant places, use these 9 values for performing comparisons
     vector<unsigned> equiSpacedVec;                             

     if(((vecSearched[0]))>vecOfVec[kbt][(vecOfVec[kbt].size())-1]) //if beginning of searched vector > last value present in individual vectors of vecOfVec then continue
     {
         continue;                             
     }                         

     unsigned elementIndex=0; //used for iterating over equiSpacedVec                             
     unsigned i=0; //used for iterating over individual buckets vecOfVec[kbt].second
     //search for value in bucket and store it in bucketValPos
     bool firstRun=true;             
     for(vector<unsigned>::iterator itValPos=vecSearched.begin();itValPos!=vecSearched.end();++itValPos)
     {
         //construct a summarized vector out of individual vectors of vecOfVec
         if(firstRun)
         {
             firstRun=false;
             unsigned elementIndex1=0; //used for iterating over equiSpacedVec
             while(elementIndex1<(vecOfVec[kbt].size())) //create a small vector for skipping over the remaining vectors
             {
                  if((elementIndex1+(10000))<(vecOfVec[kbt].size()))
                     elementIndex1+=10000;
                  else
                     break;
                 equiSpacedVec.push_back(vecOfVec[kbt][elementIndex1]);
             }          
         }
         //skip individual vectors of vecOfVec using summarized vector constructed above
         while((!(equiSpacedVec.empty()))&&(equiSpacedVec.size()>(elementIndex+1))&&((*itValPos)>equiSpacedVec[elementIndex+1])){
             elementIndex+=1;
             if((i+100)<(vecOfVec[kbt].size()))
                 i+=100;
         }            
         unsigned j=i;
         while(((*itValPos)>vecOfVec[kbt][j])&&(j<vecOfVec[kbt].size())){
             j++;
         }
         if(j>(vecOfVec[kbt].size()-1)) //element not found even at last position.
         {
             break;
         }              

         if((*itValPos)==vecOfVec[kbt][j])
         {                                     
             //store intersection result
         }
     }
  }
    return 0;
}

【问题讨论】：

你试过std::set_intersection吗？
@ixSci 我确实尝试过 set_intersection 但它给了我比我现在的代码更差的性能。
performing vector intersection in C++ 的可能重复项

标签： c++ c++11 vector

【解决方案1】：

您的问题是一个非常受欢迎的问题。由于您没有与要相交的向量相关的数据，因此归结为加快两个向量之间的交集，基本上有两种方法：

1。无需任何预处理

这通常由三件事来解决：

减少比较次数。例如，对于小向量（大小为 1 到 50），您应该对每个元素进行二进制搜索，以避免遍历主题向量的所有 9000 多个元素。

提高代码质量以减少分支错误预测，例如观察结果集通常比输入集的大小更小可以转换此类代码：

while (Apos < Aend && Bpos < Bend) {
    if (A[Apos] == B[Bpos]) {
        C[Cpos++] = A[Apos];
        Apos++; Bpos++;
    }
    else if (A[Apos] > B[Bpos]) {
        Bpos++;
    } 
    else {
      Apos++;
    } 
}

编写“展开”此类比较的代码，虽然更容易预测分支（例如块大小 = 2）：

while (1) {
    Adat0 = A[Apos]; Adat1 = A[Apos + 1]; 
    Bdat0 = B[Bpos]; Bdat1 = B[Bpos + 1];
    if (Adat0 == Bdat0) { 
        C[Cpos++] = Adat0;
    }
    else if (Adat0 == Bdat1) {
        C[Cpos++] = Adat0;
        goto advanceB;
    }
    else if (Adat1 == Bdat0) {
        C[Cpos++] = Adat1;
        goto advanceA;
    }
    if (Adat1 == Bdat1) {
        C[Cpos++] = Adat1;
        goto advanceAB;
    }
    else if (Adat1 > Bdat1) goto advanceB;
    else goto advanceA;
advanceA:
    Apos+=2;
    if (Apos >= Aend) { break; } else { continue; }
advanceB:
    Bpos+=2;
    if (Bpos >= Bend) { break; } else { continue; }
advanceAB:
    Apos+=2;  Bpos+=2;
    if (Apos >= Aend || Bpos >= Bend) { break; }
}
//  fall back to naive algorithm for remaining elements

使用 SIMD 指令执行块操作

这些技术很难在 QA 环境中描述，但您可以阅读它们（加上相关的优化，如 if conversion）here 和 here 或查找实现元素 here

2。带预处理

恕我直言，这是一个更好的方法，因为您有一个大小为 9000 多个元素的单个主题向量。你可以用它制作一个间隔树，或者简单地找到一种方法来索引它，例如制作一个可以加快搜索速度的结构：

vector<unsigned> subject(9000+); 
vector<range>    index(9000/M);

其中 range 是一个类似的结构

struct range {
    unsigned min, max; 
};

从而创建一个像这样的范围序列

[0, 100], [101, 857], ... [33221, 33500]

这将允许在进行交集时跳过许多比较（例如，如果另一个集合的元素大于子范围的最大值，您可以完全跳过该子范围）

3。并行化

是的，在两个 :P 的列表中总会有第三个元素。当您已经对过程进行了足够优化（并且仅在那时），请将您的工作分解为多个块并并行运行。这个问题符合一个令人尴尬的模式，所以 200 个向量 vs 1 应该定义为“50 vs 1 四次并发”

^{测试、测量、重新设计！！}

【讨论】：

谢谢。您能否借助示例代码解释预处理的第二种选择
@StegVerner 我相信如果你真的有实现需求，你有足够的信息可以通过谷歌搜索。如果您寻求的是解释，请查看here 第 3.1 节“通过固定宽度分区的交集”。仅这些东西是patented 的事实就足以理解实施它们不是胆小的人的努力，也不是业余时间的任务。
朴素算法涉及 (N+M) 个比较。鉴于 N ~ 9000 和 M >>> N（最多 4,000,000），我怀疑我们需要在 M 侧而不是 N 侧工作。 IE。预处理以减少 N 比较的数量可能无济于事，但在 M 端有一些向前跳过的方法将大大加快速度。当然，这取决于数据、vecVecs 中的长度分布等。我同意测试/计时是唯一的方法。

【解决方案2】：

如果我正确理解了您的代码，那么如果 N 是向量数，M 是每个向量内的元素数，那么您的算法大致为 O(N * M^2)。然后是“桶”策略，它稍微改进了一些东西，但它的效果很难一见钟情。

我建议您处理已排序的向量并在已排序的向量上进行交集。像这样的：

vector<vector<unsigned> > vecOfVec;
vector<unsignend> vecSearched ;
for (vector<unsigned> v : vecSearched) // yes, a copy
{
   std::sort(v.begin(), v.end()) ;
   if (vecSearched.empty()) // first run detection
      vSearched = v ; 
   else
   {  // compute intersection of v and vecSearch
      auto vit = v.begin() ;
      auto vend = v.end() ;
      auto sit = vecSearched.begin() ;
      auto send = vecSearched.end() ;
      vector<unsiged> result ;
      while (vit != vend && sit != send)
      {
         if (*vit < *sit)
             vit++ ;
         else if (*vit == *sit)
         {
             result.push_bck(*it) ; 
             ++vit ;
             ++sit ;
         }
         else //  *vit > *sit
            ++sit ;
      }
      vecSearched = result ;
   }
}

代码未经测试，无论如何，它背后的想法是排序向量的交集更容易，因为您可以比较这两个迭代器（vit、sit）并增长指向较小的那个。所以 intersecton 在 M 中是线性的，整个复杂度是 O(N * M *log(M))，其中 log(M) 是由于排序

【讨论】：

@maron 正如我在问题中提到的，我的向量已经排序。我不理解您提出的改进建议。你能解释一下吗？干杯
据我了解，鉴于它们都已排序，您所要做的就是找到它们重叠的第一个和最后一个，然后复制两者之间的内容。

【解决方案3】：

比较两个排序向量的最简单方法是同时遍历它们，并且只递增具有较小值的迭代器。如果两个向量都是唯一的，则可以保证需要最少的比较次数。实际上，您可以对各种集合链表使用相同的代码。

@Nikos Athanasiou（上面的答案）提供了许多有用的提示来加速您的代码，例如使用跳过列表、simd 比较。但是，您的数据集是如此之小，以至于即使是这里简单的天真代码也运行得非常快......

template<typename CONT1, typename CONT2, typename OP_MARKIDENTICAL>
inline
void set_mark_overlap(  const CONT1& cont1,
                        const CONT2& cont2,
                        OP_MARKIDENTICAL op_markidentical)
{
    auto ii = cont1.cbegin();
    auto end1 = cont1.cend();
    auto jj = cont2.cbegin();
    auto end2 = cont2.cend();

    if (cont1.empty() || cont2.empty())
        return;

    for (;;)
    {
        // increment iterator to container 1 if it is less
        if (*ii < *jj)
        {
            if (++ii == end1)
                break;
        }
        // increment iterator to container 2 if it is less
        else if (*jj < *ii)
        {
            if (++jj == end2)
                break;
        }
        // same values
        // increment both iterators
        else
        {
            op_markidentical(*ii);
            ++ii;
            if (ii == end1)
                break;
            // 
            // Comment if container1 can contain duplicates
            // 
            ++jj;
            if (jj == end2)
                break;
        }
    }
}

以下是您可以如何使用此代码：

template<typename TT>
struct op_store
{
    vector<TT>& store;
    op_store(vector<TT>& store): store(store){}
    void operator()(TT val){store.push_back(val);}
};


vector<unsigned> first{1,2,3,4,5,6};
vector<unsigned> second{1,2,5,6, 7,9};
vector<unsigned> overlap;
set_mark_overlap(   first, second, op_store<unsigned>(overlap));
for (const auto& ii : overlap)
    std::cout << ii << ",";
std::cout << ii << "\n";
// 1，2，5，6

此代码假定两个向量都不包含重复项。如果您的任何 vecOfVec 包含重复项，并且您希望打印出每个重复项，那么您需要注释上面指示的代码。如果您的 vecSearched 向量包含重复项，则不清楚什么是适当的响应...

在您的情况下，存储匹配值的代码就是这三行：

// results
vector<vector<unsigned> > results(120); 
for (unsigned ii = 0; ii < vecOfVec.size(); ++ii) 
    set_mark_overlap(vecSearched, vecOfVec[ii], op_store<unsigned>(results[ii]));

在优化方面，您的问题有两个特点： 1) 一个列表总是比另一个短得多 2) 较短的列表被重复使用，而较长的列表对于每次比较都是新的。

前期成本（预处理，例如@Nikos Athanasiou 建议的跳过列表）与 9000 的短列表（反复使用）无关，但与较长的列表无关。

我想大部分跳过是针对较长的列表，所以跳过列表可能不是灵丹妙药。当容器二赶上时（在上面的代码中），如何做一种动态跳过列表，以便您跳 N ( j += 4,000,000 / 9000) 或一个 (++j)。如果你已经跳了两步，那么你可以使用迷你二分搜索来找到合适的量来增加 j。

由于列表长度的这种不对称性，我看不出使用 SIMD 重新编码会有所帮助：我们需要将比较次数减少到 (N+M) 以下，而不是提高每次比较的速度。但是，这取决于您的数据。编写代码并计时！

这是创建一些随机数向量并检查它们是否存在的测试代码

#include <iostream>
#include <vector>
#include <unordered_set>
#include <exception>
#include <algorithm>
#include <limits>
#include <random>

using namespace std;

template<typename TT>
struct op_store
{
    std::vector<TT>& store;
    op_store(vector<TT>& store): store(store){}
    void operator()(TT val, TT val2){if (val != val2) std::cerr << val << " !- " << val2 << "\n"; store.push_back(val);}
};




void fill_vec_with_unique_random_values(vector<unsigned>& cont, unordered_set<unsigned>& curr_values)
{
    static random_device rd;
    static mt19937 e1(rd());
    static uniform_int_distribution<unsigned> uniform_dist(0, std::numeric_limits<unsigned>::max());


    for (auto& jj : cont)
    {
        for (;;) 
        {
            unsigned new_value = uniform_dist(e1);
            // make sure all values are unique
            if (curr_values.count(new_value) == 0)
            {
                curr_values.insert(new_value);
                jj = new_value;
                break;
            }
        }
    }
}

int main (int argc, char *argv[])
{
    static random_device rd;
    static mt19937 e1(rd());

    //  vector searched in contains 9000+ elements. Vectors in vecSearched are sorted
    vector<unsigned> vecSearched(9000);
    unordered_set<unsigned> unique_values_9000;
    fill_vec_with_unique_random_values(vecSearched, unique_values_9000);

    //
    //  Create 120 vectors of size ranging from 1 to 2000000 elements. All vectors in vecOfVec are sorted
    // 
    vector<vector<unsigned> > vecOfVec(5); 
    normal_distribution<> vec_size_normal_dist(1000000U, 500000U);
    for (unsigned ii = 0; ii < vecOfVec.size(); ++ii) 
    {
        std::cerr << " Create Random data set" << ii << " ...\n";
        auto vec_size = min(2000000U, static_cast<unsigned>(vec_size_normal_dist(e1)));
        vecOfVec[ii].resize(vec_size);

        // Do NOT share values with the 9000. We will manually add these later
        unordered_set<unsigned> unique_values(unique_values_9000);
        fill_vec_with_unique_random_values(vecOfVec[ii], unique_values);
    }



    // insert half of vecSearched in our 120 vectors so that we know what we are going to find
    vector<unsigned> correct_results(begin(vecSearched), begin(vecSearched) + 4500);
    for (unsigned ii = 0; ii < vecOfVec.size(); ++ii) 
        vecOfVec[ii].insert(vecOfVec[ii].end(), begin(correct_results), end(correct_results));

    // Make sure everything is sorted
    std::cerr << " Sort data ...\n";
    for (unsigned ii = 0; ii < vecOfVec.size(); ++ii) 
        sort(begin(vecOfVec[ii]), end(vecOfVec[ii]));
    sort(begin(vecSearched), end(vecSearched));
    sort(begin(correct_results), end(correct_results));


    std::cerr << " Match ...\n";
    // results
    vector<vector<unsigned> > results(120); 
    for (unsigned ii = 0; ii < vecOfVec.size(); ++ii) 
    {
        std::cerr << ii << " done\n";
        set_mark_overlap(vecSearched, vecOfVec[ii], op_store<unsigned>(results[ii]));

        // check all is well
        if (results[ii] != correct_results)
            throw runtime_error("Oops");
    }
    return(0);
}

【讨论】：

您的解决方案运行良好，但我们能否像 Nikos Athanasiou 的第 2 点那样借助预处理对其进行优化
我添加了一个关于@Nikos Athanasiou 建议的优化的简短讨论。编写备选方案、测量比较次数和多次运行时间应该很简单。我怀疑鉴于您的数据具有超级参考（向量）局部性，一些替代方案可能涉及较少的比较，但会导致更多的缓存未命中并且运行速度较慢:-(

【解决方案4】：

使用 set_intersection 可能会有所帮助，但我不知道它是否会提高整体速度：

vector<vector<unsigned int> > vecOfVec(200);
vector<unsigned int> vecSearched; 
set<unsigned int> intersection;
for(auto it = vecOfVec.begin(); it != vecOfVec.end(); ++it)
{
  std::set_intersection(it->begin(), it->end(), vecSearched.begin(), vecSearched.end(), std::inserter(intersection, intersection.begin()));
}

【讨论】：

它没有...我已经尝试过这个..正如我的 cmets 中提到的那样
哦，好吧，我没看到这条评论