使用 python 中的 csv 模块以块的形式读取和写入大型处理数据到 csv 文件答案

【问题标题】：To read and write large processed data to a csv file in chunks using csv module in python使用 python 中的 csv 模块以块的形式读取和写入大型处理数据到 csv 文件
【发布时间】：2023-01-02 20:33:02
【问题描述】：

我正在使用 csv 模块（读取）读取一个 csv 文件并将其处理为超过 5 GB 的大尺寸。然后使用 csv 模块（写入）将其写入另一个 csv。读取数据很好，但在处理数据之后.它变得超过百万条记录。所以在运行代码时它显示内存错误。所以我试图把它写成块。我没有使用 pandas，因为它很慢我在需要几天才能完成之前尝试过它..所以我正在使用 csv 模块进行读写。我不知道如何分块写入 csv 文件以避免内存问题。

我想分块写入 csv 文件。我首先读取 csv 标头并先写入它，然后将其他正文数据循环到同一个 csv 文件。如何使用 csv 模块而不是 pandas 以块的形式写入此主体数据。

import csv

def getmodified(make):
    initialize = int(make[0])
    ending = int(make[1])
    m = len(make[0])
    e = "{0:0"+str(m)+"}"
    ret = []
    for l in range(initialize,ending+1):
        ret += [e.format(l)]
    return ret
     

with open("input.csv", "r") as f:
    reader = csv.reader(f)
    writer = csv.writer(open("out.csv.csv","a", newline=""))
    header = next(reader)
    writer.writerow(header+["column_modified"])
    for r in reader:
        to_change = r[0]
        if "(" not in to_change:
            writer.writerow(r+[to_change])
        else:    
         static = to_change[:to_change.index("(")]
         make = to_change[len(static)+1:-1].split("-") 
         modify = getmodified(make)
         for each in modify:
            modified = r+[static+each]
            writer.writerow(modified)

在 getmodified(make) 函数中，我正在做范围添加所有我在将它写入 CSV 后出现错误。它能够写入 2.7GB 数据..然后它显示内存错误

input.csv

out.csv

【问题讨论】：

必须是csv吗？ Dask Dataframes（工作方式类似于 pandas）和 Parquet 文件是处理大于 RAM 工作负载的更好方法，并且针对块状读/写和多线程等进行了优化。我很高兴地使用它们来读/写 130 GB 的表格数据

标签： python python-3.x csv chunks

【解决方案1】：

将 csv 模块与 enumerate 函数结合使用。

看到这个一般的想法。

import csv

# Set the chunk size (number of rows to read/write at a time)
CHUNK_SIZE = 1000

# Open the input and output files
with open("input.csv", "r") as input_file, open("output.csv", "w") as output_file:
    # Create a CSV reader and writer
    reader = csv.reader(input_file)
    writer = csv.writer(output_file)

    # Process the data in chunks
    for i, rows in enumerate(reader, start=1):
        # Process the rows
        writer.writerows(processed_rows)

        # Print a message every CHUNK_SIZE rows
        if i % CHUNK_SIZE == 0:
            print(f"Processed {i} rows")

它将一次以 CHUNK_SIZE 行的块的形式读取和写入输入和输出文件中的数据，并在每次处理 CHUNK_SIZE 行时打印一条消息，让您了解其进度。

您可以调整块大小以满足您的需要。更大的块大小可能更有效，但也可能会消耗更多内存，并且您可能会遇到上面提到的内存错误。

【讨论】：