Python csv.DictReader 内存不足答案

【问题标题】：Python csv.DictReader out of memoryPython csv.DictReader 内存不足
【发布时间】：2019-01-04 16:01:30
【问题描述】：

我想根据时间戳对 csv 文件中的值进行排序并将其打印到另一个文件，但是对于具有多行的文件，python 会耗尽内存（当正在读取文件时）。我可以做些什么来提高效率，还是应该使用 csv.DictReader 之外的其他东西？

import csv, sys
import datetime
from pathlib import Path

localPath = "C:/MyPath"


    # data variables 
dataDir = localPath + "data/" dataExtension = ".dat" 

    pathlistData = Path(dataDir).glob('**/*'+ dataExtension)

    # Generated filename as date, Format: YYYY-DDDTHH
    generatedDataDir = localPath + "result/"
    #generatedExtension = ".dat"
    errorlog = 'errorlog.csv'

    fieldnames = ['TimeStamp', 'A', 'B', 'C', 'C', 'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L','M', 'N', 'O', 'P', 'Q', 'R'] 

    for dataPath in pathlistData:
        #stores our data in a dictionary
        dataDictionary = {}

        dataFileName = str(dataPath).replace('\\', '/')
        newFilePathString = dataFileName.replace(dataDir,generatedDataDir)

        with open(dataPath, 'r') as readFile:
            print(str("Reading data from " + dataFileName))
            keysAsDate = []#[datetime.datetime.strptime(ts, "%Y-%m-%d") for ts in timestamps]
            reader = csv.DictReader(readFile, fieldnames=fieldnames) 

            for row in reader:

                try:
                    timestamp = row['TimeStamp']
                    #create a key based on the timestamp
                    timestampKey = datetime.datetime.strptime(timestamp[0:16], "%Y-%jT%H:%M:%S")
                    #save this key as a date, used later for sorting
                    keysAsDate.append(timestampKey)
                    #save the row data in a dictionary
                    dataDictionary[timestampKey] = row

                except csv.Error as e:
                    sys.exit('file %s, line %d: %s' % (errorlog, reader.line_num, e))

            #sort the keys
            keysAsDate.sort()
        readFile.close()

        with open(newFilePathString, 'w') as writeFile:
            writer = csv.DictWriter(writeFile, fieldnames=fieldnames, lineterminator='\n')
            print(str("Writing data to " + newFilePathString))
            #loop over the sorted keys
            for idx in range(0, len(keysAsDate)):

                #get the row from our data dictionary 
                writeRow = dataDictionary[keysAsDate[idx]]
                #print(dataDictionary[keysAsDate[key]])
                writer.writerow(writeRow)
                if idx%30000 == 0:
                    print("Writing to new file: " + str(int(idx/len(keysAsDate) * 100)) + "%")


        print(str("Finished writing to file: " + newFilePathString))

        writeFile.close()

更新：我使用了 pandas 并将大文件分成更小的块，我可以单独排序。如果我将文件一个接一个地附加，这目前不能解决大量放错位置的问题。

for dataPath in pathlistData:

dataFileName = str(dataPath).replace('\\', '/')
#newFilePathString = dataFileName.replace(dataDir,generatedDataDir)


print(str("Reading data from " + dataFileName))
#divide our large data frame into smaller data frame chunks
#so we can sort the content in memory
for df_chunk in pd.read_csv(dataFileName, header = None, chunksize = chunk_size, names = fieldnames):
    dataDictionary = {}
    dataDictionary.clear()

    for idx in range(0, chunk_size):
        #print(df_chunk[idx:idx+1])
        row = df_chunk[idx:idx+1]
        dataDictionary = df_chunk.sort_values(['TimeStamp'], ascending=True)
    firstTimeStampInChunk = dataDictionary[0:1]['TimeStamp']
    #print("first: " + firstTimeStampInChunk)
    lastTimeStampInChunk = dataDictionary[chunk_size-1:chunk_size]['TimeStamp']
    #print("last: " + lastTimeStampInChunk)

    timestampStr = str(firstTimeStampInChunk)[chunk_shift:timestamp_size+chunk_shift] + str(lastTimeStampInChunk)[chunk_shift:timestamp_size+chunk_shift]
    tempFilePathString = str(timestampStr + dataExtension).replace(':', '_').replace('\\', '/')
    dataDictionary.to_csv('temp/'+tempFilePathString, header = None, index=False)

# data variables
tempDataDir = localPath + "temp/"
tempPathlistData = Path(tempDataDir).glob('**/*'+ dataExtension)

tempPathList = list(tempPathlistData)

我解决随机值问题的算法理论（无代码）是：

第 1 步 - 分成更小的块，其中“chunk_size = 内存中要处理的最大行数除以 2”

第 2 步 - 按顺序循环文件，一次合并两个文件并将它们排序在一起，然后再次拆分它们，这样就没有文件大于 chunk_size。

第 3 步 - 向后循环，一次合并两个文件并对其进行排序，然后再次拆分，因此没有文件大于 chunk_size。

第 4 步 - 现在所有严重错位的低值都应该到达最低部分，所有严重错位的高值都应该到达最高部分。按顺序追加文件！

缺点；这个时间复杂度一点也不可取，如果我没记错的话基本上是 O(N^2)

【问题讨论】：

标签： python python-3.x csv file-read

【解决方案1】：

试试 pandas csv 阅读器，效率很高。 (https://pandas.pydata.org/pandas-docs/stable/generated/pandas.read_csv.html)。您可以使用https://pandas.pydata.org/pandas-docs/stable/generated/pandas.DataFrame.to_dict.html在 pandas 和字典之间轻松转换

【讨论】：

尽管 pandas 可能更高效，但遗憾的是它并不能解决这个问题。内存根本无法一次读取内存中的所有值。但是我可以使用 chunk_size 将文件划分为较小的文件并对它们进行排序。但是，对于大量偏移的值，它并不能完全解决问题。
您是否尝试过增加分配给 SWAP 内存的大小？它会很慢，但至少你不会耗尽内存

【解决方案2】：

您解释说内存排序不适合您，因为文件大小超过了内存大小。至少有两种方法可以解决这个问题。两者都依赖于执行更多的文件 I/O。

将长记录压缩到一个内存效率高的文件偏移量。在读取每条记录（或对行长求和）时调用tell()，并在内存中仅保留时间戳和文件偏移量。按时间戳对偏移量进行排序。在您浏览已排序的元组时反复调用 seek()，随机读取一条记录，并将其附加到您的输出文件中。
更好的方法是让/usr/bin/sort 进行外部合并排序。 Windows 用户可以从https://git-scm.com/download/ 获取coreutils GNU sort。使用subprocess 模块调用它。

【讨论】：