c++ - 在 C++ 中处理非常大的数据

Question

我有一个 .csv包含 ~3GB 数据的文件。我想读取所有数据并进行处理。以下程序从文件中读取数据并将其存储到 std::vector<std::vector<std::string>> 中.但是，程序运行时间过长导致应用程序(vscode)卡顿，需要重启。我做错了什么？

#include <algorithm>
#include <iostream>
#include <fstream>
#include "sheet.hpp"

extern std::vector<std::string> split(const std::string& str, const std::string& delim);

int main() {
    Sheet sheet;
    std::ifstream inputFile;
    inputFile.open("C:/Users/1032359/cpp-projects/Straggler Job Analyzer/src/part-00001-of-00500.csv");
    std::string line;
    while(inputFile >> line) {
        sheet.addRow(split(line, ","));
    }
    return 0;
}

//split和 Sheet的成员函数已经过彻底测试并且工作正常。 split虽然有 N^2 的复杂性......

---

EDIT1:已根据评论中的建议修复读取的文件。

拆分函数:

std::vector<std::string> split(const std::string& str, const std::string& delim) {
    std::vector<std::string> vec_of_tokens;
    std::string token;
    for (auto character : str) {
        if (std::find(delim.begin(), delim.end(), character) != delim.end()) {
            vec_of_tokens.push_back(token);
            token = "";
            continue;
        }
        token += character;
    }
    vec_of_tokens.push_back(token);
    return vec_of_tokens;
}

---

编辑2:虚拟 csv 行: 5612000000,5700000000,4665712499,798,3349189123,0.02698,0.06714,0.07715,0.004219,0.004868,0.06726,7.915e-05,0.0003681,0.27,0.00293,3.285,0.008261,0,0,0.01608

limits:  
field1: starting timestamp (nanosecs)
field2: ending timestamp (nanosecs)
field3: job id (<= 1,000,000)
field4: task id (<= 10,000)
field5: machine id (<= 30,000,000)
field6: CPU time (sorry, no clue)
field7-20: no idea, unused for the current stage, but needed for later stages.

---

EDIT3:需要的输出

还记得 Excel 中的 .thenby 函数吗？
这里的排序顺序是首先在第 5 列(基于 1 的索引)排序，然后在第 3 列，最后在第 4 列；全部上升。

Answer 1

我会首先定义一个类来承载有关一条记录的信息，并为 operator>> 添加重载。和 operator<<帮助从/向流读取/写入记录。我可能还会添加一个帮助程序来处理逗号分隔符。

首先，我使用的 header 集:

#include <algorithm>   // sort
#include <array>       // array
#include <cstdint>     // integer types
#include <filesystem>  // filesystem
#include <fstream>     // ifstream
#include <iostream>    // cout
#include <iterator>    // istream_iterator
#include <tuple>       // tie
#include <vector>      // vector

一个简单的定界符助手可能如下所示。它丢弃 ( ignore() ) 分隔符，如果它在流中或设置 failbit如果分隔符不存在，则在流上。

template <char Char> struct delimiter {};

template <char Char> // read a delimiter
std::istream& operator>>(std::istream& is, const delimiter<Char>) {
    if (is.peek() == Char) is.ignore();
    else is.setstate(std::ios::failbit);
    return is;
}

template <char Char> // write a delimiter
std::ostream& operator<<(std::ostream& os, const delimiter<Char>) {
    return os.put(Char);
}

实际record根据您提供的信息，类可以如下所示:

struct record {
    uint64_t start;       // ns
    uint64_t end;         // ns
    uint32_t job_id;      // [0,1000000]
    uint16_t task_id;     // [0,10000]
    uint32_t machine_id;  // [0,30000000]
    double cpu_time;
    std::array<double, 20 - 6> unknown;
};

然后可以使用 delimiter 从流中读取这样的记录类模板(实例化为使用逗号和换行符作为分隔符):

std::istream& operator>>(std::istream& is, record& r) {
    delimiter<','> del;
    delimiter<'\n'> nl;

    // first read the named fields
    if (is >> r.start >> del >> r.end >> del >> r.job_id >> del >>
        r.task_id >> del >> r.machine_id >> del >> r.cpu_time)
    {
        // then read the unnamed fields:
        for (auto& unk : r.unknown) is >> del >> unk;
    }
    return is >> nl;
}

写入记录的方式类似:

std::ostream& operator<<(std::ostream& os, const record& r) {
    delimiter<','> del;
    delimiter<'\n'> nl;

    os << 
        r.start << del <<
        r.end << del <<
        r.job_id << del <<
        r.task_id << del <<
        r.machine_id << del <<
        r.cpu_time;
    for(auto&& unk : r.unknown) os << del << unk;
    return os << nl;
}

将整个文件读入内存，排序并打印结果:

int main() {
    std::filesystem::path filename = "C:/Users/1032359/cpp-projects/"
                                     "Straggler Job Analyzer/src/part-00001-of-00500.csv";
    
    std::vector<record> records;

    // Reserve space for "3GB" / 158 (the length of a record + some extra bytes)
    // records. Increase the 160 below if your records are actually longer on average:
    records.reserve(std::filesystem::file_size(filename) / 160);

    // open the file
    std::ifstream inputFile(filename);

    // copy everything from the file into `records`
    std::copy(std::istream_iterator<record>(inputFile),
              std::istream_iterator<record>{},
              std::back_inserter(records));

    // sort on columns 5-3-4 (ascending)
    auto sorter = [](const record& lhs, const record& rhs) {
        return std::tie(lhs.machine_id, lhs.job_id, lhs.task_id) <
               std::tie(rhs.machine_id, rhs.job_id, rhs.task_id);
    };
    std::sort(records.begin(), records.end(), sorter);

    // display the result
    for(auto& r : records) std::cout << r;
}

在我的带有旋转磁盘的旧计算机上，上述过程大约需要 2 分钟。如果这太慢，我会测量长时间运行的部分的时间:

• reserve
• copy
• sort

然后，您可能可以使用该信息来尝试找出需要改进的地方。例如，如果排序有点慢，使用 std::vector<double> 可能会有所帮助而不是 std::array<double, 20-6>存储未命名的字段:

struct record {
    record() : unknown(20-6) {} 

    uint64_t start;       // ns
    uint64_t end;         // ns
    uint32_t job_id;      // [0,1000000]
    uint16_t task_id;     // [0,10000]
    uint32_t machine_id;  // [0,30000000]
    double cpu_time;
    std::vector<double> unknown;
};