Viết dữ liệu vào đĩa bằng Python như một quá trình nền

Question

tôi có một chương trình bằng Python rằng về cơ bản nào sau đây:

for j in xrange(200): 
    # 1) Compute a bunch of data 
    # 2) Write data to disk 

1) mất khoảng 2-5 phút
2) mất khoảng ~ 1 phút

Lưu ý rằng có quá nhiều dữ liệu cần lưu trong bộ nhớ.

Lý tưởng nhất là những gì tôi muốn làm là ghi dữ liệu vào đĩa theo cách tránh không tải CPU. Điều này có thể thực hiện được bằng Python không? Cảm ơn!

Answer 1

Bạn có thể thử using multiple processes như thế này:

import multiprocessing as mp 

def compute(j): 
    # compute a bunch of data 
    return data 

def write(data): 
    # write data to disk 

if __name__ == '__main__': 
    pool = mp.Pool() 
    for j in xrange(200): 
     pool.apply_async(compute, args=(j,), callback=write) 
    pool.close() 
    pool.join() 

pool = mp.Pool() sẽ tạo ra một hồ bơi của các quá trình lao động. Theo mặc định, số lượng công nhân bằng số lõi CPU mà máy của bạn có.

Mỗi hàng đợi pool.apply_async một nhiệm vụ được điều hành bởi một nhân viên trong nhóm quy trình công nhân. Khi một nhân viên có mặt, nó hoạt động compute(j). Khi công nhân trả về một giá trị, data, một luồng trong quy trình chính chạy hàm gọi lại write(data), với data là dữ liệu được trả về bởi người lao động.

Một số hãy cẩn thận:

• Các dữ liệu này phải được picklable, vì nó đã được truyền đạt từ quá trình lao động trở lại quá trình chính thông qua một Queue.
• Không đảm bảo rằng thứ tự mà người lao động hoàn thành nhiệm vụ giống như thứ tự mà tác vụ được gửi đến hồ bơi . Vì vậy, thứ tự dữ liệu được ghi vào đĩa có thể không tương ứng với j từ 0 đến 199. Một cách xung quanh vấn đề này sẽ ghi dữ liệu vào cơ sở dữ liệu sqlite (hoặc loại khác) với j của các trường dữ liệu. Sau đó, khi bạn muốn đọc dữ liệu theo thứ tự, bạn có thể SELECT * FROM table ORDER BY j.

• Sử dụng nhiều quy trình sẽ tăng dung lượng bộ nhớ cần thiết khi dữ liệu được tạo bởi các quy trình công nhân và dữ liệu đang chờ ghi vào đĩa tích lũy trong Hàng đợi. Bạn có thể giảm dung lượng bộ nhớ cần thiết bằng cách sử dụng các mảng NumPy . Nếu điều đó là không thể, sau đó bạn có thể phải giảm bớt số của các quá trình:

  pool = mp.Pool(processes=1) 
  

  Điều đó sẽ tạo ra một quá trình lao động (để chạy compute), để lại quá trình chính để chạy write. Vì compute mất nhiều thời gian hơn write, Hàng đợi sẽ không được sao lưu với nhiều hơn một đoạn dữ liệu được ghi vào đĩa.Tuy nhiên, bạn vẫn cần đủ bộ nhớ để tính toán trên một đoạn dữ liệu trong khi viết một đoạn dữ liệu khác nhau của vào đĩa.

  Nếu bạn không có đủ bộ nhớ để thực hiện đồng thời, bạn không có lựa chọn nào - mã ban đầu của bạn, chạy theo thứ tự compute và write, là cách duy nhất.

Answer 2

Bạn có thể sử dụng một cái gì đó như Queue.Queue (module là ở đây: Queue) và threading.Thread (hoặc threading.start_new_thread nếu bạn chỉ muốn có một chức năng), các mô-đun là ở đây: threading - Là một tập tin ghi không phải là CPU thâm canh và sử dụng nhiều IO. (và GIL không ảnh hưởng đến nó).

Answer 3

Cách đơn giản là chỉ sử dụng luồng và hàng đợi. Mặt khác, nếu phần tính toán không phụ thuộc vào trạng thái toàn cầu, và bạn có máy với nhiều lõi CPU, cách hiệu quả hơn sẽ được sử dụng process pool

from multiprocessing import Pool 

def compute_data(x): 
    return some_calculation_with(x) 

if __name__ == '__main__': 
    pool = Pool(processes=4) # let's say you have quad-core, so start 4 workers 

    with open("output_file","w") as outfile: 
     for calculation_result in pool.imap(compute_data, range(200)): 
     # pool.imap returns results as they come from process pool  
      outfile.write(calculation_result)