python非阻塞写入csv文件

Question

我正在编写一些 python 代码来进行一些计算并将结果写入文件。这是我当前的代码：

for name, group in data.groupby('Date'):
    df = lot_of_numpy_calculations(group)

    with open('result.csv', 'a') as f:
        df.to_csv(f, header=False, index=False)

计算和写入有时都需要。我读了一些关于 python 中的异步的文章，但我不知道如何实现它。有没有一种简单的方法来优化这个循环，使它不会等到写入完成并开始下一次迭代？

Answer 1

由于 numpy 和 pandas io 都不支持 asyncio，因此对于线程而言，这可能是比 asyncio 更好的用例。 （此外，基于 asyncio 的解决方案无论如何都会在幕后使用线程。）

例如，此代码生成一个编写器线程，您使用队列向其提交工作：

import threading, queue

to_write = queue.Queue()

def writer():
    # Call to_write.get() until it returns None
    for df in iter(to_write.get, None):
        with open('result.csv', 'a') as f:
            df.to_csv(f, header=False, index=False)
threading.Thread(target=writer).start()

for name, group in data.groupby('Date'):
    df = lot_of_numpy_calculations(group)
    to_write.put(df)
# enqueue None to instruct the writer thread to exit
to_write.put(None)

请注意，如果写入结果始终比计算慢，则队列将不断累积数据帧，这最终可能会消耗大量内存。在这种情况下，请务必通过将 maxsize 参数传递给 constructor 来为队列提供最大大小。

另外，请考虑为每次写入重新打开文件会减慢写入速度。如果写入的数据量很小，或许可以通过提前打开文件来获得更好的性能。

Answer 2

由于大多数操作系统don't support异步文件I/O，现在常见的跨平台方法是使用线程。

例如，aiofiles 模块封装了线程池来为 asyncio 提供文件 I/O API。