如何使用循环或并行计算将大数据输入 python pandas？答案

【问题标题】：How to input large data into python pandas using looping or parallel computing?如何使用循环或并行计算将大数据输入 python pandas？
【发布时间】：2017-12-10 07:08:47
【问题描述】：

我有一个 8gb 的 csv 文件，我无法运行代码，因为它显示内存错误。

file = "./data.csv"
df = pd.read_csv(file, sep="/", header=0, dtype=str)

我想使用 python 将文件分成 8 个小文件（“按 id 排序”）。最后，有一个循环，以便输出文件将有所有 8 个文件的输出。

或者我想尝试并行计算。主要目标是在 python pandas 中处理 8gb 数据。谢谢。

我的 csv 文件包含大量以“/”作为逗号分隔符的数据，

id    venue           time             code    value ......
AAA   Paris      28/05/2016 09:10      PAR      45   ......
111   Budapest   14/08/2016 19:00      BUD      62   ......
AAA   Tokyo      05/11/2016 23:20      TYO      56   ......
111   LA         12/12/2016 05:55      LAX      05   ......
111   New York   08/01/2016 04:25      NYC      14   ......
AAA   Sydney     04/05/2016 21:40      SYD      2    ......
ABX   HongKong   28/03/2016 17:10      HKG      5    ......
ABX   London     25/07/2016 13:02      LON      22   ......
AAA   Dubai      01/04/2016 18:45      DXB      19   ......
.
.
.
.

【问题讨论】：

使用 itertools 作为答案这里解释stackoverflow.com/questions/16289859/…
您真的需要这 8 个小文件还是只使用最终文件？
只有最终文件
@Iris 所以基本上你想按 id 对你的csv 进行排序并将其保存到文件中？

标签： python loops csv pandas parallel-processing

【解决方案1】：

pandas read_csv 有两个参数选项，您可以使用它们来做您想做的事情：

nrows : to specify the number of rows you want to read
skiprows : to specify the first row you want to read

请参阅文档：https://pandas.pydata.org/pandas-docs/stable/generated/pandas.read_csv.html

【讨论】：

【解决方案2】：

您可能还想使用 das 框架，它内置在 dask.dataframe 中。从本质上讲，csv 文件被转换为多个 pandas 数据帧，每个数据帧都在必要时读取。但是，并不是每个 pandas 命令都可以在 dask 中使用。

【讨论】：

【解决方案3】：

使用chunksize 参数一次读取一个块并将文件保存到磁盘。这会将原始文件分成等份，每份 100000 行：

file = "./data.csv"
chunks = pd.read_csv(file, sep="/", header=0, dtype=str, chunksize = 100000)

for it, chunk in enumerate(chunks):
    chunk.to_csv('chunk_{}.csv'.format(it), sep="/")

如果您知道原始文件的行数，您可以计算出精确的chunksize 以将文件分成 8 等份 (nrows/8)。

【讨论】：

虽然在迭代保存之前加载了整个数据帧，但这是否仍然会消耗太多内存？
没有。分块的全部意义在于它不会将整个数据帧加载到内存中。我的答案中的变量chunks 是一个可迭代的对象，它几乎不占用内存（在此处阅读更多内容）[pandas.pydata.org/pandas-docs/stable/io.html#io-chunking].只有当您遍历 chunks 时，您才会真正将文件的块大小版本读入内存。

【解决方案4】：

import numpy as np
from multiprocessing import Pool

def processor(df):

    # Some work

    df.sort_values('id', inplace=True)
    return df

size = 8
df_split = np.array_split(df, size)

cores = 8
pool = Pool(cores)
for n, frame in enumerate(pool.imap(processor, df_split), start=1):
    frame.to_csv('{}'.format(n))
pool.close()
pool.join()

【讨论】：

嘿！这很酷！！我正在寻找类似的东西！但我得到这个错误，frame.to_csv(output, sep="^", index=False.format(n)) AttributeError: 'bool' object has no attribute 'format'
哪里，输出 = "/file.csv"
frame.to_csv(output, sep="^", index=False)
文件“/usr/lib/python2.7/multiprocessing/pool.py”，第 659 行，在下一个提升值 IndexError：位置索引器超出范围
你的处理器功能里面有什么？

【解决方案5】：

如果您不需要所有列，也可以使用usecols 参数：

https://pandas.pydata.org/pandas-docs/stable/generated/pandas.read_csv.html

usecols : array-like or callable, default None

Return a subset of the columns. [...] 
Using this parameter results in much faster parsing time and lower memory usage.

【讨论】：