在 16 核机器上流式传输大型 (5gb) CSV 的 FAST（并行？）？答案

【问题标题】：Streaming Large (5gb) CSV’s FAST (in parallel?) on 16-core Machine?在 16 核机器上流式传输大型 (5gb) CSV 的 FAST（并行？）？
【发布时间】：2016-03-12 21:11:11
【问题描述】：

我有 30 个 CSV 文件（另存为 .txt 文件），每个 2GB 到 11GB 不等，位于一台 16 核的服务器上。

每个 CSV 的每一行都包含一个日期、一个时间和一个 ID。

我需要构建一个大小为 datetime x ID（大约 35,000 x 2000）的密集矩阵，其中每个单元格都是具有此 datetime 和 ID 的行数（因此每个 CSV 行的 datetime 和 ID 都用作矩阵索引来更新这个矩阵）。每个文件都包含一个唯一的日期时间范围，因此这项工作在文件之间是并行的，令人尴尬。

问题：什么是更快/最快的方法来完成这个并（可能）并行化它？我偏爱 Python，但如果有的话可以在 C++ 中工作那里有更好的解决方案。我应该用 MapReduce 还是 MPI 重写？看看 Dask 还是 Pandas？以某种方式编译我的python脚本？完全不同的东西？

我目前的方法（我很乐意放弃它以换取更快的方法）： 目前，我在 Python 中连续执行此操作（一次一个 CSV）并将输出矩阵保存为 h5 格式。我使用以下命令从命令行逐行流式传输 CSV：

cat one_csv.txt | my_script.py > outputfile.h5

我的 python 脚本的工作方式如下：

# initialize matrix
…
for line in sys.stdin:
    # Split the line into data columns
    split =  line.replace('\n','').split(',')
    ...(extract & process datetime; extract ID)...

    # Update matrix
    matrix[datetime, ID] = matrix[datetime, ID] +1

编辑下面是 CSV 文件中的几行示例。唯一相关的列是“dateYMDD”（格式化为“80101”表示 2008 年 1 月 1 日）、“时间”和“ID”。因此，例如，代码应该读取使用下面 CSV 的第一行将 1 添加到对应于 (Jan_1_2008_00_00_00, 12) 的矩阵单元中。

另外：唯一时间比唯一 ID 多得多，并且 CSV 是按时间排序的。

Type|Number|dateYMDD|time|ID
2|519275|80101|0:00:00|12
5|525491|80101|0:05:00|25
2|624094|80101|0:12:00|75
5|623044|80102|0:01:00|75
6|658787|80102|0:03:00|4

【问题讨论】：

您需要从分析开始。例如 - 如果由于大量竞争性并行读取而浪费大量时间在磁盘上寻找，则可能会获得更差的性能。这些天对 SSD 来说还不错，但您必须考虑整个系统在处理过程中如何交互。
你能分享几行CSV文件和你想要什么矩阵吗？我可以尝试用 python map/reduce 编写它。
正如@JohnLaRooy 所说，此任务是磁盘绑定的，并行化会使事情变得更糟，除非您使用多个磁盘（但这也会使 PCI 总线饱和）。
能否考虑在写入csv文件的同时创建h5文件，然后合并？
添加到原始帖子的示例 CSV。伊夫，我不听从你在创建 csv 文件的同时创建 h5 文件的建议——你能澄清一下吗？我已经拥有所有的 CSV 文件。我使用 h5 作为存储脚本输出矩阵的便捷方式。 -感谢cmets！

标签： python csv optimization parallel-processing bigdata

【解决方案1】：

首先，您可能应该分析您的脚本，以确保瓶颈确实在您的想法中。

也就是说，Python 的 Global Interpreter Lock 将使并行化变得困难，除非您使用多处理，而且我希望单独处理它们并合并结果会更快：为每个 Python 脚本提供一个 CSV 并输出到一个表，然后合并表。如果表格比 CSV 小得多（如果单元格具有高值，则可以预期），那么这应该是相对有效的。

不过，我认为这不会让你像你提到的那样全速全速前进。如果这不符合您的期望，我会考虑用 C++、Rust 或 Cython 编写它。

【讨论】：