读取、格式化然后写入大型 CSV 文件答案

【问题标题】：Read, format, then write large CSV files读取、格式化然后写入大型 CSV 文件
【发布时间】：2017-10-22 16:56:57
【问题描述】：

我有相当大的 csv 文件，我需要逐行操作/修改（因为每一行可能需要不同的修改规则），然后将它们以正确的格式写入另一个 csv。

目前，我有：

import multiprocessing

def read(buffer):
    pool = multiprocessing.Pool(4)
    with open("/path/to/file.csv", 'r') as f:
        while True:
            lines = pool.map(format_data, f.readlines(buffer))
            if not lines:
                break
            yield lines

def format_data(row):
    row = row.split(',') # Because readlines() returns a string
    # Do formatting via list comprehension
    return row

def main():
    buf = 65535
    rows = read(buf)
    with open("/path/to/new.csv",'w') as out:
        writer = csv.writer(f, lineterminator='\n')
        while rows:
            try:
                writer.writerows(next(rows))
            except StopIteration:
                break

即使我通过map 使用多处理并通过生成器防止内存过载，处理 40,000 行仍然需要 2 分钟以上。老实说，它不应该花那么多钱。我什至从生成器输出中生成了一个嵌套列表，并尝试将数据一次写入一个大文件，这是一种逐块的方法，但仍然需要很长时间。我在这里做错了什么？

【问题讨论】：

标签： python-3.x csv multiprocessing generator bigdata

【解决方案1】：

我想通了。

首先，问题出在我的format_data() 函数中。它正在调用数据库连接，每次运行时，它都会构建数据库连接并在每次迭代时关闭它。

我通过字典创建基本映射来修复它，以实现支持多线程的指数级快速查找表。

所以，我的代码如下所示：

import multiprocessing

def read(buffer):
    pool = multiprocessing.Pool(4)
    with open("/path/to/file.csv", 'r') as f:
        while True:
            lines = pool.map(format_data, f.readlines(buffer))
            if not lines:
                break
            yield lines

def format_data(row):
    row = row.split(',') # Because readlines() returns a string
    # Do formatting via list comprehension AND a dictionary lookup
    # vice a database connection
    return row

def main():
    rows = read(1024*1024)
    with open("/path/to/new.csv",'w') as out:
        while rows:
            try:
                csv.writer(f, lineterminator='\n').writerows(next(rows))
            except StopIteration:
                break

我能够在不到 30 秒的时间内解析约 150MB 的文件。在这里吸取的一些经验教训可供其他人借鉴。

【讨论】：