Giải quyết các vấn đề song song khó xử bằng cách sử dụng đa xử lý Python

Question

Làm thế nào để sử dụng multiprocessing để xử lý embarrassingly parallel problems?

vấn đề Embarassingly song song thường bao gồm ba phần cơ bản:

dữ liệu đầu vào

1. đọc (từ một tập tin, cơ sở dữ liệu, kết nối tcp, vv).
2. Chạy các phép tính trên dữ liệu đầu vào, trong đó mỗi phép tính là độc lập với bất kỳ phép tính nào khác.
3. Viết kết quả tính toán (vào tệp, cơ sở dữ liệu, kết nối tcp, v.v.).

Chúng ta có thể parallelize chương trình theo hai chiều:

• Phần 2 có thể chạy trên nhiều lõi, vì mỗi tính là độc lập; thứ tự xử lý không quan trọng.
• Mỗi phần có thể chạy độc lập. Phần 1 có thể đặt dữ liệu trên hàng đợi đầu vào, phần 2 có thể kéo dữ liệu ra khỏi hàng đợi đầu vào và đưa kết quả vào hàng đợi đầu ra, và phần 3 có thể kéo kết quả ra khỏi hàng đợi đầu ra và ghi chúng ra.

này dường như là một mô hình cơ bản nhất trong lập trình đồng thời, nhưng tôi vẫn đang bị mất trong cố gắng giải quyết nó, vì vậy chúng ta hãy viết một ví dụ kinh điển để minh họa cách này được thực hiện sử dụng đa xử.

Đây là vấn đề ví dụ: Cho một số CSV file với các hàng số nguyên làm đầu vào, tính tổng của chúng. Tách vấn đề thành ba phần, tất cả có thể chạy song song:

1. Process file đầu vào dữ liệu thô (danh sách/iterables các số nguyên)
2. Tính số tiền của dữ liệu, song song
3. Output số tiền

Dưới đây là truyền thống, đơn quá trình ràng buộc chương trình Python mà giải quyết ba nhiệm vụ:

#!/usr/bin/env python 
# -*- coding: UTF-8 -*- 
# basicsums.py 
"""A program that reads integer values from a CSV file and writes out their 
sums to another CSV file. 
""" 

import csv 
import optparse 
import sys 

def make_cli_parser(): 
    """Make the command line interface parser.""" 
    usage = "\n\n".join(["python %prog INPUT_CSV OUTPUT_CSV", 
      __doc__, 
      """ 
ARGUMENTS: 
    INPUT_CSV: an input CSV file with rows of numbers 
    OUTPUT_CSV: an output file that will contain the sums\ 
"""]) 
    cli_parser = optparse.OptionParser(usage) 
    return cli_parser 


def parse_input_csv(csvfile): 
    """Parses the input CSV and yields tuples with the index of the row 
    as the first element, and the integers of the row as the second 
    element. 

    The index is zero-index based. 

    :Parameters: 
    - `csvfile`: a `csv.reader` instance 

    """ 
    for i, row in enumerate(csvfile): 
     row = [int(entry) for entry in row] 
     yield i, row 


def sum_rows(rows): 
    """Yields a tuple with the index of each input list of integers 
    as the first element, and the sum of the list of integers as the 
    second element. 

    The index is zero-index based. 

    :Parameters: 
    - `rows`: an iterable of tuples, with the index of the original row 
     as the first element, and a list of integers as the second element 

    """ 
    for i, row in rows: 
     yield i, sum(row) 


def write_results(csvfile, results): 
    """Writes a series of results to an outfile, where the first column 
    is the index of the original row of data, and the second column is 
    the result of the calculation. 

    The index is zero-index based. 

    :Parameters: 
    - `csvfile`: a `csv.writer` instance to which to write results 
    - `results`: an iterable of tuples, with the index (zero-based) of 
     the original row as the first element, and the calculated result 
     from that row as the second element 

    """ 
    for result_row in results: 
     csvfile.writerow(result_row) 


def main(argv): 
    cli_parser = make_cli_parser() 
    opts, args = cli_parser.parse_args(argv) 
    if len(args) != 2: 
     cli_parser.error("Please provide an input file and output file.") 
    infile = open(args[0]) 
    in_csvfile = csv.reader(infile) 
    outfile = open(args[1], 'w') 
    out_csvfile = csv.writer(outfile) 
    # gets an iterable of rows that's not yet evaluated 
    input_rows = parse_input_csv(in_csvfile) 
    # sends the rows iterable to sum_rows() for results iterable, but 
    # still not evaluated 
    result_rows = sum_rows(input_rows) 
    # finally evaluation takes place as a chain in write_results() 
    write_results(out_csvfile, result_rows) 
    infile.close() 
    outfile.close() 


if __name__ == '__main__': 
    main(sys.argv[1:]) 

Hãy lấy chương trình này và viết lại nó để sử dụng đa xử lý để song song ba phần được nêu ở trên. Dưới đây là bộ xương của mới chương trình này, song song, mà cần phải được fleshed ra để giải quyết những phần trong các ý kiến:

#!/usr/bin/env python 
# -*- coding: UTF-8 -*- 
# multiproc_sums.py 
"""A program that reads integer values from a CSV file and writes out their 
sums to another CSV file, using multiple processes if desired. 
""" 

import csv 
import multiprocessing 
import optparse 
import sys 

NUM_PROCS = multiprocessing.cpu_count() 

def make_cli_parser(): 
    """Make the command line interface parser.""" 
    usage = "\n\n".join(["python %prog INPUT_CSV OUTPUT_CSV", 
      __doc__, 
      """ 
ARGUMENTS: 
    INPUT_CSV: an input CSV file with rows of numbers 
    OUTPUT_CSV: an output file that will contain the sums\ 
"""]) 
    cli_parser = optparse.OptionParser(usage) 
    cli_parser.add_option('-n', '--numprocs', type='int', 
      default=NUM_PROCS, 
      help="Number of processes to launch [DEFAULT: %default]") 
    return cli_parser 


def main(argv): 
    cli_parser = make_cli_parser() 
    opts, args = cli_parser.parse_args(argv) 
    if len(args) != 2: 
     cli_parser.error("Please provide an input file and output file.") 
    infile = open(args[0]) 
    in_csvfile = csv.reader(infile) 
    outfile = open(args[1], 'w') 
    out_csvfile = csv.writer(outfile) 

    # Parse the input file and add the parsed data to a queue for 
    # processing, possibly chunking to decrease communication between 
    # processes. 

    # Process the parsed data as soon as any (chunks) appear on the 
    # queue, using as many processes as allotted by the user 
    # (opts.numprocs); place results on a queue for output. 
    # 
    # Terminate processes when the parser stops putting data in the 
    # input queue. 

    # Write the results to disk as soon as they appear on the output 
    # queue. 

    # Ensure all child processes have terminated. 

    # Clean up files. 
    infile.close() 
    outfile.close() 


if __name__ == '__main__': 
    main(sys.argv[1:]) 

Những mẩu mã, cũng như another piece of code that can generate example CSV files cho mục đích thử nghiệm, có thể found on github.

Tôi sẽ đánh giá cao bất kỳ thông tin chi tiết nào ở đây về cách thức mà bạn rất kinh nghiệm đồng thời sẽ tiếp cận vấn đề này.

---

Dưới đây là một số câu hỏi tôi có khi nghĩ về vấn đề này. điểm thưởng để giải quyết bất kỳ/tất cả:

• Tôi có nên có các quy trình con để đọc trong các dữ liệu và đặt nó vào hàng đợi, hoặc quá trình chính có thể làm điều này mà không ngăn chặn cho đến khi tất cả các đầu vào được đọc?
• Tương tự như vậy, tôi có nên có một quy trình con để viết kết quả ra khỏi hàng đợi đã xử lý hay không, quá trình chính có thực hiện được điều này mà không phải đợi tất cả kết quả không?
• Tôi có nên sử dụng processes pool cho tổng số hoạt động không?
  • Nếu có, tôi sẽ gọi phương thức nào trên hồ để bắt đầu xử lý kết quả vào hàng đợi nhập, mà không chặn quá trình nhập và xuất? apply_async()? map_async()? imap()? imap_unordered()?
• Giả sử chúng ta không cần phải ngắt hàng đợi đầu vào và đầu ra như dữ liệu đã nhập, nhưng có thể đợi cho đến khi tất cả đầu vào được phân tích và tất cả kết quả được tính toán (ví dụ, vì chúng ta biết tất cả đầu vào và đầu ra sẽ phù hợp với bộ nhớ hệ thống). Chúng tôi có nên thay đổi thuật toán theo bất kỳ cách nào (ví dụ: không chạy bất kỳ quy trình nào đồng thời với I/O) không?

Answer 1

Giải pháp của tôi có thêm chuông và còi để đảm bảo rằng thứ tự của đầu ra có cùng thứ tự của đầu vào. Tôi sử dụng multiprocessing.queue để gửi dữ liệu giữa các tiến trình, gửi các thông báo dừng để mỗi quá trình biết thoát khỏi việc kiểm tra các hàng đợi. Tôi nghĩ rằng các ý kiến ​​trong nguồn nên làm cho nó rõ ràng những gì đang xảy ra nhưng nếu không cho tôi biết.

#!/usr/bin/env python 
# -*- coding: UTF-8 -*- 
# multiproc_sums.py 
"""A program that reads integer values from a CSV file and writes out their 
sums to another CSV file, using multiple processes if desired. 
""" 

import csv 
import multiprocessing 
import optparse 
import sys 

NUM_PROCS = multiprocessing.cpu_count() 

def make_cli_parser(): 
    """Make the command line interface parser.""" 
    usage = "\n\n".join(["python %prog INPUT_CSV OUTPUT_CSV", 
      __doc__, 
      """ 
ARGUMENTS: 
    INPUT_CSV: an input CSV file with rows of numbers 
    OUTPUT_CSV: an output file that will contain the sums\ 
"""]) 
    cli_parser = optparse.OptionParser(usage) 
    cli_parser.add_option('-n', '--numprocs', type='int', 
      default=NUM_PROCS, 
      help="Number of processes to launch [DEFAULT: %default]") 
    return cli_parser 

class CSVWorker(object): 
    def __init__(self, numprocs, infile, outfile): 
     self.numprocs = numprocs 
     self.infile = open(infile) 
     self.outfile = outfile 
     self.in_csvfile = csv.reader(self.infile) 
     self.inq = multiprocessing.Queue() 
     self.outq = multiprocessing.Queue() 

     self.pin = multiprocessing.Process(target=self.parse_input_csv, args=()) 
     self.pout = multiprocessing.Process(target=self.write_output_csv, args=()) 
     self.ps = [ multiprocessing.Process(target=self.sum_row, args=()) 
         for i in range(self.numprocs)] 

     self.pin.start() 
     self.pout.start() 
     for p in self.ps: 
      p.start() 

     self.pin.join() 
     i = 0 
     for p in self.ps: 
      p.join() 
      print "Done", i 
      i += 1 

     self.pout.join() 
     self.infile.close() 

    def parse_input_csv(self): 
      """Parses the input CSV and yields tuples with the index of the row 
      as the first element, and the integers of the row as the second 
      element. 

      The index is zero-index based. 

      The data is then sent over inqueue for the workers to do their 
      thing. At the end the input process sends a 'STOP' message for each 
      worker. 
      """ 
      for i, row in enumerate(self.in_csvfile): 
       row = [ int(entry) for entry in row ] 
       self.inq.put((i, row)) 

      for i in range(self.numprocs): 
       self.inq.put("STOP") 

    def sum_row(self): 
     """ 
     Workers. Consume inq and produce answers on outq 
     """ 
     tot = 0 
     for i, row in iter(self.inq.get, "STOP"): 
       self.outq.put((i, sum(row))) 
     self.outq.put("STOP") 

    def write_output_csv(self): 
     """ 
     Open outgoing csv file then start reading outq for answers 
     Since I chose to make sure output was synchronized to the input there 
     is some extra goodies to do that. 

     Obviously your input has the original row number so this is not 
     required. 
     """ 
     cur = 0 
     stop = 0 
     buffer = {} 
     # For some reason csv.writer works badly across processes so open/close 
     # and use it all in the same process or else you'll have the last 
     # several rows missing 
     outfile = open(self.outfile, "w") 
     self.out_csvfile = csv.writer(outfile) 

     #Keep running until we see numprocs STOP messages 
     for works in range(self.numprocs): 
      for i, val in iter(self.outq.get, "STOP"): 
       # verify rows are in order, if not save in buffer 
       if i != cur: 
        buffer[i] = val 
       else: 
        #if yes are write it out and make sure no waiting rows exist 
        self.out_csvfile.writerow([i, val]) 
        cur += 1 
        while cur in buffer: 
         self.out_csvfile.writerow([ cur, buffer[cur] ]) 
         del buffer[cur] 
         cur += 1 

     outfile.close() 

def main(argv): 
    cli_parser = make_cli_parser() 
    opts, args = cli_parser.parse_args(argv) 
    if len(args) != 2: 
     cli_parser.error("Please provide an input file and output file.") 

    c = CSVWorker(opts.numprocs, args[0], args[1]) 

if __name__ == '__main__': 
    main(sys.argv[1:]) 

Answer 2

Trường học cũ.

p1.py

import csv 
import pickle 
import sys 

with open("someFile", "rb") as source: 
    rdr = csv.reader(source) 
    for line in eumerate(rdr): 
     pickle.dump(line, sys.stdout) 

p2.py

import pickle 
import sys 

while True: 
    try: 
     i, row = pickle.load(sys.stdin) 
    except EOFError: 
     break 
    pickle.dump(i, sum(row)) 

p3.py

import pickle 
import sys 
while True: 
    try: 
     i, row = pickle.load(sys.stdin) 
    except EOFError: 
     break 
    print i, row 

Dưới đây là cấu trúc đa xử lý cuối cùng.

python p1.py | python p2.py | python p3.py 

Có, vỏ đã đan chúng lại với nhau ở cấp hệ điều hành. Nó có vẻ đơn giản với tôi và nó hoạt động rất độc đáo.

Có, có thêm một chút chi phí khi sử dụng dưa (hoặc cPickle). Việc đơn giản hóa, tuy nhiên, có vẻ đáng để nỗ lực.

Nếu bạn muốn tên tệp là đối số cho p1.py, đó là một thay đổi dễ dàng.

Quan trọng hơn, một chức năng như sau rất tiện dụng.

def get_stdin(): 
    while True: 
     try: 
      yield pickle.load(sys.stdin) 
     except EOFError: 
      return 

đó cho phép bạn làm điều này:

for item in get_stdin(): 
    process item 

---

này rất đơn giản, nhưng nó không dễ dàng cho phép bạn có nhiều bản sao của P2.py chạy.

Bạn có hai vấn đề: fan-out và fan-in. P1.py phải bằng cách nào đó quạt ra nhiều P2.py. Và P2.py phải bằng cách nào đó kết hợp các kết quả của chúng thành một P3.py duy nhất.

Phương pháp tiếp cận trường học cũ để làm nổi bật là một kiến ​​trúc "Đẩy" rất hữu ích.

Về mặt lý thuyết, nhiều P2.py kéo từ hàng đợi thông thường là phân bổ nguồn tài nguyên tối ưu. Điều này thường lý tưởng, nhưng nó cũng là một số lượng công bằng của lập trình. Lập trình có thực sự cần thiết không? Hay chế biến vòng tròn sẽ đủ tốt?

Thực tế, bạn sẽ thấy rằng làm cho P1.py thực hiện giao dịch "vòng tròn" đơn giản giữa nhiều P2.py có thể khá tốt. Bạn đã có cấu hình P1.py để xử lý các bản sao của P2.py qua các đường ống có tên là n. P2.py sẽ đọc từ ống thích hợp của họ.

Điều gì sẽ xảy ra nếu một trong P2.py nhận được tất cả dữ liệu "trường hợp xấu nhất" và chạy phía sau? Vâng, round-robin không hoàn hảo. Nhưng nó tốt hơn chỉ một P2.py và bạn có thể giải quyết sự thiên vị này với sự ngẫu nhiên đơn giản.

Việc gắn kết từ nhiều P2.py đến một P3.py vẫn phức tạp hơn một chút. Tại thời điểm này, cách tiếp cận trường học cũ dừng lại là thuận lợi. P3.py cần phải đọc từ nhiều ống được đặt tên bằng cách sử dụng thư viện select để xen kẽ các lần đọc.

Answer 3

Có thể giới thiệu một chút về tính song song vào phần 1. Có thể không phải là vấn đề với định dạng đơn giản như CSV, nhưng nếu việc xử lý dữ liệu đầu vào chậm hơn đáng kể so với việc đọc dữ liệu, bạn có thể đọc các khối lớn hơn, sau đó tiếp tục đọc cho đến khi tìm thấy "dấu tách hàng" (dòng mới trong trường hợp CSV, nhưng lại phụ thuộc vào định dạng đã đọc; không hoạt động nếu định dạng đủ phức tạp).

Những đoạn này, mỗi phần có thể chứa nhiều mục, sau đó có thể được tạo thành một đám đông các quy trình song song đọc các công việc ở hàng đợi, nơi chúng được phân tích cú pháp và tách, sau đó được đặt vào hàng đợi cho giai đoạn 2.

Answer 4

Tôi nhận ra rằng tôi hơi muộn cho bữa tiệc, nhưng gần đây tôi đã phát hiện ra GNU parallel và muốn thể hiện mức độ dễ dàng để hoàn thành nhiệm vụ điển hình này với nó.

cat input.csv | parallel ./sum.py --pipe > sums 

Something như thế này sẽ làm cho sum.py:

#!/usr/bin/python 

from sys import argv 

if __name__ == '__main__': 
    row = argv[-1] 
    values = (int(value) for value in row.split(',')) 
    print row, ':', sum(values) 

Parallel sẽ chạy sum.py cho mỗi dòng trong input.csv (song song, tất nhiên), sau đó xuất kết quả để sums. Rõ ràng tốt hơn so với multiprocessing rắc rối

Answer 5

Đến muộn để đảng ...

joblib có một lớp trên đầu trang của đa để giúp làm song song cho vòng lặp. Nó cung cấp cho bạn các tiện ích như việc gửi đi công việc lười biếng và báo cáo lỗi tốt hơn ngoài cú pháp rất đơn giản của nó.

Là tuyên bố từ chối trách nhiệm, tôi là tác giả ban đầu của joblib.