那些遇到过的问题

C++ 对于大型数据集（数百万行）最好的排序容器和方法是什么

S..*_*S.. 5 c++ sorting algorithm containers dataset

我正在处理一个练习，该练习应该准确地对此类代码的时间复杂度进行基准测试。

我正在处理的数据由像这样的字符串对组成hbFvMF,PZLmRb，每个字符串在数据集中出现两次，一次在位置1，一次在位置2。所以第一个字符串将指向zvEcqe,hbFvMF例如并且列表还在继续......

50k 对的示例数据集

我已经能够生成代码，对最多 50k 对的数据集进行排序没有太大问题，大约需要 4-5 分钟。10k 只需几秒钟即可完成排序。

问题是我的代码应该处理最多 500 万对的数据集。所以我想看看我还能做些什么。我将发布我的两个最佳尝试，第一个是向量，我认为我可以通过替换来升级，vector因为unsorted_map搜索时的时间复杂度更好，但令我惊讶的是，当我测试它时，两个容器之间几乎没有区别。我不确定我解决问题的方法或我选择的容器是否导致了排序时间过长......

尝试使用向量：

#include <iostream>
#include <fstream>
#include <string>
#include <sstream>
#include <map>
#include <unordered_map>
#include <stdio.h>
#include <vector>
#include <iterator>
#include <utility>
#include <functional>
#include <algorithm>

using namespace std;


template<typename T>
void search_bricks_backwards(string resume, vector<T>& vec, vector<string>& vec2) {
    int index = 0;
    int temp_index = 0;
    while (true) {
        if (index == vec.size()) {
            vec2.insert(vec2.begin(), vec[temp_index].first); 
            cout << "end of backward search, exitting..." << endl;
            break;


        }
        
        if (vec[index].second == resume) {
            vec2.insert(vec2.begin(), resume);
            

            resume = vec[index].first;
            //vec.erase(vec.begin() + index);
            temp_index = index;

            index = 0;
        }
        
        index++;
    }

}


template<typename T>
void search_bricks(string start, vector<T>& vec, vector<string>& vec2) {
    int index = 0;
    int temp_index = 0;
    while (true) {
        //cout << "iteration " << index << endl;
        if (index == vec.size()) {
            vec2.push_back(vec[temp_index].second);
            
            cout << "all forward bricks sorted" << endl;
            break;


        }
        if (vec[index].first == start) {
            vec2.push_back(vec[index].first);
            
            
            start = vec[index].second;
            //vec.erase(vec.begin() + index);
            temp_index = index;
            index = 0;
            
        }
        
        index++;
    }

    search_bricks_backwards(vec[0].first, vec, vec2);

}

template<typename T>
void search_bricks_recursion(string start, vector<T>& vec, vector<string>& vec2) {
    int index = 0;
    for (const auto& pair : vec) {
        //cout << "iteration " << index << endl;
        if (pair.first == start) {
            vec2.push_back(start);
            cout << "found " << start << " and " << pair.first << endl;
            search_bricks(pair.second, vec, vec2);
        }
        if (index + 1 == vec.size()) {
            search_bricks_backwards(start, vec, vec2);
            

        }
        index++;
    }
    
}

template<typename T>
void printVectorElements(vector<T>& vec)
{
    for (auto i = 0; i < vec.size(); ++i) {
        cout << "(" << vec.at(i).first << ","
            << vec.at(i).second << ")" << " ";
    }
    cout << endl;
}

vector<string> split(string s, string delimiter) {
    size_t pos_start = 0, pos_end, delim_len = delimiter.length();
    string token;
    vector<string> res;

    while ((pos_end = s.find(delimiter, pos_start)) != string::npos) {
        token = s.substr(pos_start, pos_end - pos_start);
        pos_start = pos_end + delim_len;
        res.push_back(token);
    }

    res.push_back(s.substr(pos_start));
    return res;
}

unordered_map<string, string> brick_to_map(string const& s)
{
    unordered_map<string, string> m;

    string key, val;
    istringstream iss(s);

    while (getline(getline(iss, key, ',') >> ws, val))
        m[key] = val;

    return m;
}

int main()
{
    vector<pair<string, string>> bricks;
    
    vector<string> sorted_bricks;

    ifstream inFile;
    inFile.open("input-pairs-50K.txt"); //open the input file

    stringstream strStream;
    strStream << inFile.rdbuf(); //read the file
    string str = strStream.str(); //str holds the content of the file

    //cout << str << endl;
    
    istringstream iss(str);
    
    for (string line; getline(iss, line); )
    {
     
        string delimiter = ",";
        string s = line;
        vector<string> v = split(s, delimiter);
        string s1 = v.at(0);
        string s2 = v.at(1);
        

        bricks.push_back(make_pair(s1, s2));
    }

   
    search_bricks(bricks[0].second, bricks, sorted_bricks);
    
    
    //display the results
    for (auto i = sorted_bricks.begin(); i != sorted_bricks.end(); ++i)
        cout << *i << " ";


    
 
}
Run Code Online (Sandbox Code Playgroud)

尝试unsorted map：

#include <iostream>
#include <fstream>
#include <string>
#include <sstream>
#include <map>
#include <unordered_map>
#include <stdio.h>
#include <vector>
#include <iterator>
#include <utility>
#include <functional>
#include <algorithm>

using namespace std;


void search_bricks_backwards(string resume, unordered_map<string, string> brick_map, vector<string>& vec2) {
    
    typedef unordered_map<string, string>::value_type map_value_type;
    while (true) {

        unordered_map<string, string>::const_iterator got = find_if(brick_map.begin(), brick_map.end(), [&resume](const map_value_type& vt)
            { return vt.second == resume; }
        );
        if (got == brick_map.end()) {
            vec2.insert(vec2.begin(), resume); 
            cout << "end of backward search, exitting..." << endl;
            break;


        }
        //cout << "iteration " << index << endl;
        else if (got->second == resume) {
            vec2.insert(vec2.begin(), resume);

            
            resume = got->first;
        
        }

       
    }

}


void search_bricks(string start, unordered_map<string, string> brick_map, vector<string>& vec2) {
    
    typedef unordered_map<string, string>::value_type map_value_type;
    while (true) {
        

        unordered_map<string, string>::const_iterator got = find_if(brick_map.begin(), brick_map.end(), [&start](const map_value_type& vt)
            { return vt.first == start; }
        );
        if (got == brick_map.end()) {
            vec2.push_back(start);

            cout << "all forward bricks sorted" << endl;
            
            break;
        }
        else if (got->first == start) {
            vec2.push_back(start);

            //cout << "found " << start << " and " << vec[index].first << endl;
            start = got->second;
            
        }
    }
    auto it = brick_map.begin();
    search_bricks_backwards(it->first, brick_map, vec2);
    


    

    

}


template<typename T>
void printVectorElements(vector<T>& vec)
{
    for (auto i = 0; i < vec.size(); ++i) {
        cout << "(" << vec.at(i).first << ","
            << vec.at(i).second << ")" << " ";
    }
    cout << endl;
}

vector<string> split(string s, string delimiter) {
    size_t pos_start = 0, pos_end, delim_len = delimiter.length();
    string token;
    vector<string> res;

    while ((pos_end = s.find(delimiter, pos_start)) != string::npos) {
        token = s.substr(pos_start, pos_end - pos_start);
        pos_start = pos_end + delim_len;
        res.push_back(token);
    }

    res.push_back(s.substr(pos_start));
    return res;
}


int main()
{
    unordered_map<string, string> bricks;
    
    vector<string> sorted_bricks;

    ifstream inFile;
    inFile.open("input-pairs-50K.txt"); //open the input file

    for (string line; getline(inFile, line); )
    {

        string delimiter = ",";
        string s = line;
        vector<string> v = split(s, delimiter);
        string s1 = v.at(0);
        string s2 = v.at(1);


        bricks.insert(make_pair(s1, s2));
    }


    /*for (auto& x : bricks)
        std::cout << x.first << "," << x.second << " ";*/


    auto it = bricks.begin();
    search_bricks(it->second, bricks, sorted_bricks);


    // display results
    for (auto i = sorted_bricks.begin(); i != sorted_bricks.end(); ++i)
        cout << *i << " ";




}
Run Code Online (Sandbox Code Playgroud)

我希望提高代码的时间复杂度，以便能够更有效地处理数据，如果有人可以建议我的代码或容器方面需要改进什么，我将非常感激。

Who*_*aig 3

首先，以目的为导向，类比一下这里真正所做的事情。这个问题有时会出现在面试问题中。它通常被表述为一组巴士票：

您正走在街上，手里拿着厚厚一叠巴士票，准备旋风般地游览周边国家。每张票都有一个出发城市和一个结束城市。你不小心把票掉在地上，它们就被吹得到处都是。你拿起它们，但随后发现它们坏了。您的任务是将它们按顺序放回原处，这样您就不必每次需要使用下一张票时都搜索堆栈。

例如，其中Sn是 s 公交车站 ID，表示从一个车站到另一个车站 的Sn->Sm行程。给出以下四张车票，涵盖五个车站（排名不分先后）：SnSm

S4->S2
S1->S5
S3->S4
S5->S3
Run Code Online (Sandbox Code Playgroud)

正确的顺序可以这样认为：

S1->S5
    S5->S3
        S3->S4
            S4->S2
Run Code Online (Sandbox Code Playgroud)

因此，正确的“排序”顺序是

S1
S5
S3
S4
S2
Run Code Online (Sandbox Code Playgroud)

算法分析

  • 算法中最大的杀手是那些序列扫描。地图提供关键的查找操作。地图find成员是让他们打勾（并发光）的原因。如果你想发出这种尖叫声，你需要使用键控来做到这一点。
  • 紧随其后的是惩罚性的顺序扫描，这是预先挂起的高昂费用std::vector，它的速度可能会变得异常昂贵。向量是连续的连续存储。附加到它们并没有那么糟糕，因为它们的子分配器往往会在扩展时过度分配，以便在必须重新分配之前为更多的推迟留出空间。但在他们之前是可怕的。

我扫描了您提供的源测试。数据中实际上有 49999 行，而不是 50000 行。经过一番尝试、错误、抓耳挠腮等之后，我发现了这一点：

  • bWyUVV仅出现一次，作为左侧。
  • EZkYGM文件中仅出现一次，作为右侧

这些必须是排序列表的终结符。如果一切顺利并且数据是原始的，则最终列表将以 开头bWyUVV，并以 结尾EZkYGM对编写算法没有帮助，但绝对有助于验证我们做了正确的事情。

全方位的性能改进

从中剥离大量代码，同时仍然保留基本前提，请考虑以下事项：

#include <iostream>
#include <fstream>
#include <string>
#include <sstream>
#include <unordered_map>
#include <deque>
#include <algorithm>
#include <chrono>

void search_bricks_backwards
(
    std::string resume, 
    std::unordered_map<std::string, std::string>& rbricks, 
    std::deque<std::string>& dst
) 
{
    while (true) 
    {
        auto it = rbricks.find(resume);
        dst.emplace_front(std::move(resume));

        if (it == rbricks.end()) 
            break;

        resume = it->second;
        rbricks.erase(it);
    }
}

void search_bricks
(
    std::unordered_map<std::string, std::string>& bricks, 
    std::unordered_map<std::string, std::string>& rbricks, 
    std::deque<std::string>& dst
) 
{
    // remember these
    std::string start = bricks.begin()->second;
    std::string resume = bricks.begin()->first;

    while (true) 
    {
        auto it = bricks.find(start);
        dst.emplace_back(std::move(start));

        if (it == bricks.end()) 
            break;

        start = it->second;
        bricks.erase(it);
    }

    // same search, but different keyed map
    search_bricks_backwards(resume, rbricks, dst);
}


int main()
{
    using namespace std::chrono;

    std::unordered_map<std::string, std::string> bricks, rbricks;
    std::deque<std::string> sorted_bricks;

    std::ifstream inFile("input-pairs-50K.txt");
    if (inFile.is_open())
    {
        for (std::string line; std::getline(inFile, line);)
        {
            // make damn sure we get two keys
            std::istringstream iss(line);
            std::string s1, s2;
            if (std::getline(iss, s1, ',') && 
                std::getline(iss, s2) &&
                !s1.empty() &&
                !s2.empty())
            {
                bricks.emplace(std::make_pair(s1, s2));
                rbricks.emplace(std::make_pair(s2, s1));
            }
        }

        auto tp0 = high_resolution_clock::now();
        search_bricks(bricks, rbricks, sorted_bricks);
        auto tp1 = high_resolution_clock::now();

        std::cout << duration_cast<milliseconds>(tp1-tp0).count() << "ms\n";

        // display results
        int n = 0;
        for (auto i = sorted_bricks.begin(); i != sorted_bricks.end(); ++i)
            std::cout << ++n << ". " << *i << '\n';
    }
}
Run Code Online (Sandbox Code Playgroud)

最显着的差异：

  • 用作std::deque<std::string>排序结果。您为这些矢量展示位置付出了高昂的代价。所有向量操作要么在后面推（对于保留向量来说可以），要么在前面推（向量中的性能很糟糕）。它std::deque专门用于非常快速的前后插入和修剪。我们不做后者，但正在大力做前者。

  • 两个映射，分别键控 s1==>s2 和 s2==>s1。有第三方容器可以为您执行此操作（例如：boost::bimap）。然而，为此，考虑到密钥大小，仅存储两个映射很容易。

  • 在适当的地方使用移动语义（尽管不多）

  • 在搜索过程中，每个发现的键都会从刚刚搜索的地图中删除。这确保我们在应该停止的时候停止，并相当快地恢复内存占用。

  • 适用时参考参数。这些容器在构建结果集时被有效地销毁，但没关系（实际上可以保证正确检测一端或另一端的终端）。

双键操作带来了天壤之别。使用您的测试数据集在一台四年前的小型 MacBook Pro 笔记本电脑上构建的发布版本会产生以下结果（您可以通过已知答案测试来验证）。

所有代码均使用 -O2 优化构建，运行 Apple clang 版本 13.0.0 (clang-1300.0.29.30)

std::unordered_map

34ms
1. bWyUVV
2. mPBGRC
3. WfkvWy
4. vjWNHY
5. HudtyD
6. DhxjdV
7. kdWhGX
8. tIsDXh
9. eMVeMX
10. fVQoeG

... 49980 lines omitted ...

49990. YfBDnP
49991. sHKVrT
49992. ZzhoZV
49993. Dyunmj
49994. KCQbpj
49995. rbMSgD
49996. WKOksU
49997. qqMTnq
49998. llrqUI
49999. XYpBnk
50000. EZkYGM
Run Code Online (Sandbox Code Playgroud)

尽管在我的测试中存在高达 42 毫秒和低至 28 毫秒的异常值，但该时序数字相当一致。使用常规线束的相同线束std::map会产生：

std::map

92ms
1. bWyUVV
2. mPBGRC
3. WfkvWy
4. vjWNHY
5. HudtyD
6. DhxjdV
7. kdWhGX
8. tIsDXh
9. eMVeMX
10. fVQoeG

... 49980 lines omitted ...

49990. YfBDnP
49991. sHKVrT
49992. ZzhoZV
49993. Dyunmj
49994. KCQbpj
49995. rbMSgD
49996. WKOksU
49997. qqMTnq
49998. llrqUI
49999. XYpBnk
50000. EZkYGM
Run Code Online (Sandbox Code Playgroud)

因此，您肯定使用了具有无序键控的正确容器，只是您实际上并未将键控用于算法的重要部分，从而使容器基本上不比顺序扫描更好。这与您的评估相符，它确实比矢量顺序扫描解决方案好不了多少；不管怎样，你基本上都是这么做的。

我怀疑上面显示的性能应该能够运行您预期的 500 万对，只要您可以将其全部保留在内存中（两次，抱歉，但结果非常坦率地表明它值得这个价格）。

更内存友好（至少一点）

下面的代码执行相同的操作，但没有添加所有输出，并使用单个被调用函数，并在飞行中转置地图。这从一开始就减少了对两张地图的需求。整体性能与上面的代码进行比较；只是一个不同的、更内存友好的实现。

#include <iostream>
#include <fstream>
#include <algorithm>
#include <string>
#include <sstream>
#include <deque>
#include <unordered_map>
#include <chrono>

using map_type = std::unordered_map<std::string, std::string>;

std::deque<std::string> sort_bricks( map_type& bricks )
{
    std::deque<std::string> dst;

    // remember these
    std::string start = bricks.begin()->second;
    std::string resume = bricks.begin()->first;

    // process by append
    while (true) 
    {
        auto it = bricks.find(start);
        dst.emplace_back(std::move(start));

        if (it == bricks.end()) 
            break;

        start = std::move(it->second);
        bricks.erase(it);
    }

    // invert the remaining map
    {
        map_type mres;
        for (auto& pr : bricks)
            mres.emplace(std::move(pr.second), pr.first);
        std::swap(bricks, mres);
    }
    
    // process by prepend
    while (true) 
    {
        auto it = bricks.find(resume);
        dst.emplace_front(std::move(resume));

        if (it == bricks.end()) 
            break;

        resume = std::move(it->second);
        bricks.erase(it);
    }

    return dst;
}

int main()
{
    using namespace std::chrono;

    std::ifstream inFile("input-pairs-50K.txt");
    if (inFile.is_open())
    {
        map_type bricks;
        for (std::string line; std::getline(inFile, line);)
        {
            std::istringstream iss(line);
            std::string s1, s2;
            if (std::getline(iss, s1, ',') && 
                std::getline(iss, s2) &&
                !s1.empty() &&
                !s2.empty())
            {
                bricks.emplace(std::make_pair(s1, s2));
            }
        }

        auto tp0 = high_resolution_clock::now();
        auto sorted_bricks = sort_bricks(bricks);
        auto tp1 = high_resolution_clock::now();

        std::cout << "Size: " << sorted_bricks.size() << '\n';
        std::cout << "Time: " << duration_cast<milliseconds>(tp1-tp0).count() << "ms\n";
    }
}
Run Code Online (Sandbox Code Playgroud)

归档时间：

查看次数：

446 次

最近记录：

3 年，6 月 前
相关归档
为什么我们不能声明一个std :: vector <AbstractClass>？ 78
R的排名和顺序 43
Quickselect的平均运行时间 27
如何在动态图中避免"堆指针意大利面"？ 27
这意味着什么是const int*&var？ 25
一些数字之间最大的GCD 11
寻找最小的pangrammatic窗口的有效算法？ 7
移动初始化列表的元素是否安全？ 5
如何对类似于链表的条目进行排序？ 5
带百分比的随机数 0
难疑归档
使用Git将特定文件重置或还原到特定版本？ 4255
在一行中捕获多个异常(块除外) 2521
在JavaScript中创建多行字符串 2412
什么是MVP和MVC,有什么区别？ 2081
如何更改一个特定提交的提交作者？ 1949
如果目录尚不存在,如何mkdir？ 1784
什么是非捕获组？(？:)做什么？ 1653
C#中两个问号共同意味着什么？ 1540
创建一个带参数的Bash别名？ 1164
有效地使用Git和Dropbox？ 1117