逐行读取csv文件并保存满足某些条件的行

Question

我有一个已经在多个主题中讨论过的问题，但我想更深入一点，也许会找到更好的解决方案。

所以我们的想法是用 python 遍历“巨大的”（50 到 60GB）.csv 文件，找到满足某些条件的行，提取它们，最后将它们存储在第二个变量中以供进一步分析。

最初的问题在于 r 脚本，我使用 sparklyr 连接进行管理，或者最终在 bash 中使用一些 gawk 代码（请参阅 awk 或 gawk）来提取我需要的数据，然后使用 R/python 对其进行分析。

我想专门用 python 来解决这个问题，这个想法是避免混合使用 bash/python 或 bash/R (unix) 等语言。到目前为止，我使用 open as x，并逐行浏览文件，它有点工作，但它非常慢。例如，通过文件非常快（每秒约 500.000 行，即使对于 58M 行也可以），但是当我尝试存储数据时，速度下降到每秒约 10 行。对于大约 300.000 行的提取，这是不可接受的。

我尝试了几种解决方案，但我猜这不是最优的（糟糕的 python 代码？:( ）最终存在更好的解决方案。

解决方案 1： 遍历文件，将行拆分为列表，检查条件，如果可以，则将满足条件的每次迭代的行放入 numpy 矩阵和 vstack 中（非常慢）

import csv
import numpy
import pandas
from tqdm import tqdm

date_first = '2008-11-01'
date_last = '2008-11-10'

a = numpy.array(['colnames']*35) #data is 35 columns
index = list()

with open("data.csv", "r") as f:
    for line in tqdm(f, unit = " lines per"):
        line = line.split(sep = ";") # csv with ";" ...
        date_file = line[1][0:10] # date stored in the 2nd column 

        if  date_file >= date_first and date_file <= date_last : #data extraction concern a time period (one month for example)
            line=numpy.array(line) #go to numpy
            a=numpy.vstack((a, line)) #stack it

解决方案 2： 相同，但如果条件正常（非常慢），则将该行存储在带有行索引的 pandas data.frame 中

import csv
import numpy
import pandas
from tqdm import tqdm

date_first = '2008-11-01'
date_last = '2008-11-10'
row = 0 #row index
a = pandas.DataFrame(numpy.zeros((0,35)))#data is 35 columns

with open("data.csv", "r") as f:
    for line in tqdm(f, unit = " lines per"):
        line = line.split(sep = ";")
        date_file = line[1][0:10]

        if  date_file>=date_first and date_file<=date_last :
            a.loc[row] = line #store the line in the pd.data.frame at the position row
            row = row + 1 #go to next row

解决方案 3： 相同，但不是将行存储在某处，这对我来说是主要问题，而是保留满足行的索引，然后用我需要的行打开 csv （更慢，实际上通过文件查找索引已经足够快了，打开索引的行非常慢）

import csv
import numpy
import pandas
from tqdm import tqdm

date_first = '2008-11-01'
date_last = '2008-11-10'
row = 0
index = list()

with open("data.csv", "r") as f:
    f = csv.reader(f, delimiter = ";")
    for line in tqdm(f, unit = " lines per"):
        line = line.split(sep = ";")
        date_file = line[1][0:10]
        row = row + 1
        if  date_file>=date_first and date_file<=date_last :
            index.append(row)

with open("data.csv") as f:
    reader=csv.reader(f)
    interestingrows=[row for idx, row in enumerate(reader) if idx in index]

这个想法是只保留满足条件的数据，这里是特定月份的提取。我不明白问题出在哪里，将数据保存在某处（vstack，或在 pd.DF 中写入）绝对是一个问题。我很确定我做错了什么，但我不确定在哪里/什么。

数据是一个包含 35 列和超过 5700 万行的 csv。 感谢阅读

O.

Answer 1

追加到数据帧和 numpy 数组非常昂贵，因为每次追加都必须将整个数据复制到新的内存位置。相反，您可以尝试分块读取文件，处理数据，然后追加。在这里，我选择了 100,000 的块大小，但您显然可以更改它。

我不知道你的 CSV 的列名，所以我猜是'date_file'。这应该让你接近：

import pandas as pd

date_first = '2008-11-01'
date_last = '2008-11-10'

df = pd.read_csv("data.csv", chunksize=100000)

for chunk in df:
    chunk = chunk[(chunk['date_file'].str[:10] >= date_first)
                  & (chunk['date_file'].str[:10] <= date_last)]
    chunk.to_csv('output.csv', mode='a')