Pertanyaan Memecahkan masalah paralel yang mengherankan menggunakan multiprocessing Python


Bagaimana cara orang menggunakannya multiprocessing untuk mengatasi masalah paralel yang memalukan?

Masalah paralel yang memusingkan biasanya terdiri dari tiga bagian dasar:

  1. Baca baca input data (dari file, database, koneksi tcp, dll.).
  2. Menjalankan perhitungan pada data input, di mana setiap kalkulasi dilakukan independen dari perhitungan lainnya.
  3. Menulis hasil perhitungan (ke file, database, koneksi tcp, dll.).

Kami dapat memparalelkan program dalam dua dimensi:

  • Bagian 2 dapat berjalan di beberapa core, karena setiap perhitungan bersifat independen; urutan pemrosesan tidak masalah.
  • Setiap bagian dapat berjalan secara mandiri. Bagian 1 dapat menempatkan data pada antrian input, bagian 2 dapat menarik data dari antrian input dan memasukkan hasil ke antrian output, dan bagian 3 dapat menarik hasil dari antrian output dan menulisnya.

Ini tampaknya merupakan pola paling dasar dalam pemrograman konkuren, tetapi saya masih bingung dalam memecahkannya, jadi mari tulis contoh kanonik untuk mengilustrasikan bagaimana hal ini dilakukan dengan menggunakan multiprocessing.

Berikut ini contoh masalah: Diberikan a File CSV dengan deretan bilangan bulat sebagai input, hitung jumlahnya. Pisahkan masalah menjadi tiga bagian, yang semuanya bisa berjalan secara paralel:

  1. Memproses file input menjadi data mentah (daftar / iterables dari bilangan bulat)
  2. Hitung jumlah data, secara paralel
  3. Output jumlahnya

Di bawah ini adalah program python tradisional dengan proses tunggal yang menyelesaikan tiga tugas ini:

#!/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:])

Mari kita ambil program ini dan menulis ulang untuk menggunakan multiprocessing untuk memparalelkan ketiga bagian yang diuraikan di atas. Di bawah ini adalah kerangka program paralel baru ini, yang perlu disempurnakan untuk membahas bagian-bagian dalam komentar:

#!/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:])

Potongan-potongan kode ini, juga potongan kode lain yang dapat menghasilkan file CSV contoh untuk tujuan pengujian, bisa ditemukan di github.

Saya sangat menghargai wawasan di sini tentang bagaimana Anda ahli konkurensi akan mendekati masalah ini.


Inilah beberapa pertanyaan yang saya miliki ketika memikirkan masalah ini. Poin bonus untuk menangani setiap / semua:

  • Haruskah saya memiliki proses anak untuk membaca data dan menempatkannya ke dalam antrian, atau dapatkah proses utama melakukan ini tanpa memblokir sampai semua masukan dibaca?
  • Demikian juga, haruskah saya memiliki proses anak untuk menulis hasil dari antrian yang diproses, atau dapatkah proses utama melakukan ini tanpa harus menunggu semua hasil?
  • Haruskah saya menggunakan proses pool untuk operasi penjumlahan?
  • Anggap kita tidak perlu menyedot antrian input dan output ketika data dimasukkan, tetapi bisa menunggu hingga semua input diuraikan dan semua hasil dihitung (misalnya, karena kita tahu semua input dan output akan masuk dalam memori sistem). Haruskah kita mengubah algoritme dengan cara apa pun (mis., Tidak menjalankan proses bersamaan dengan I / O)?

76
2018-03-01 21:38


asal


Jawaban:


Solusi saya memiliki bel dan peluit ekstra untuk memastikan bahwa urutan output sama dengan urutan input. Saya menggunakan multiprocessing.queue untuk mengirim data antar proses, mengirim pesan berhenti sehingga setiap proses mengetahui untuk berhenti memeriksa antrean. Saya pikir komentar di sumber harus menjelaskan apa yang terjadi tetapi jika tidak memberi tahu saya.

#!/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:])

64
2018-03-02 16:16



Saya menyadari bahwa saya agak terlambat untuk pesta, tetapi saya baru saja menemukan Paralel GNU, dan ingin menunjukkan betapa mudahnya menyelesaikan tugas khas ini dengannya.

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

Sesuatu seperti ini akan dilakukan 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)

Paralel akan berjalan sum.py untuk setiap baris dalam input.csv (secara paralel, tentu saja), kemudian output hasilnya sums. Jelas lebih baik daripada multiprocessing merepotkan


5
2017-08-23 11:00



Datang terlambat ke pesta ...

joblib memiliki lapisan di atas multiprocessing untuk membantu membuat paralel untuk loop. Ini memberi Anda fasilitas seperti pengiriman pekerjaan yang malas, dan pelaporan kesalahan yang lebih baik di samping sintaks yang sangat sederhana.

Sebagai disclaimer, saya adalah penulis asli dari joblib.


5
2018-01-15 08:13



Sekolah Tua.

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

Inilah struktur akhir multi-processing.

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

Ya, shell telah merajut ini bersama-sama di tingkat OS. Kelihatannya lebih sederhana bagi saya dan ini bekerja dengan sangat baik.

Ya, ada sedikit lebih banyak overhead dalam menggunakan acar (atau cPickle). Penyederhanaan, bagaimanapun, tampaknya sepadan dengan usaha.

Jika Anda ingin nama file menjadi argumen p1.py, itu perubahan yang mudah.

Lebih penting lagi, fungsi seperti berikut ini sangat berguna.

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

Itu memungkinkan Anda melakukan ini:

for item in get_stdin():
     process item

Ini sangat sederhana, tetapi tidak dengan mudah memungkinkan Anda memiliki banyak salinan P2.py yang berjalan.

Anda memiliki dua masalah: kipas dan fan-in. The P1.py entah bagaimana harus menyebar ke beberapa P2.py. Dan P2.py itu entah bagaimana harus menggabungkan hasil mereka menjadi P3.py. tunggal.

Pendekatan sekolah tua untuk fan-out adalah arsitektur "Push", yang sangat efektif.

Secara teoritis, beberapa P2.py menarik dari antrian umum adalah alokasi sumber daya yang optimal. Ini sering ideal, tetapi juga cukup banyak pemrograman. Apakah pemrogramannya benar-benar diperlukan? Atau akankah proses round-robin menjadi cukup baik?

Secara praktis, Anda akan menemukan bahwa membuat P1.py melakukan "round robin" sederhana dalam menangani beberapa P2.py mungkin cukup bagus. Anda harus mengatur P1.py untuk mengatasinya n salinan P2.py melalui pipa bernama. The P2.py masing-masing akan membaca dari pipa yang sesuai.

Bagaimana jika satu P2.py mendapatkan semua data "kasus terburuk" dan berjalan di belakang? Ya, round-robin tidak sempurna. Tapi itu lebih baik daripada hanya satu P2.py dan Anda dapat mengatasi bias ini dengan pengacakan sederhana.

Fan-in dari beberapa P2.py untuk satu P3.py sedikit lebih rumit, masih. Pada titik ini, pendekatan sekolah lama berhenti menjadi menguntungkan. P3.py perlu membaca dari beberapa pipa bernama menggunakan select perpustakaan untuk interleave membaca.


4
2018-03-01 21:55



Mungkin juga memperkenalkan sedikit paralelisme ke dalam bagian 1. Mungkin tidak masalah dengan format yang sesederhana CSV, tetapi jika pemrosesan input data terasa lebih lambat daripada pembacaan data, Anda bisa membaca potongan yang lebih besar, kemudian terus membaca sampai Anda menemukan "pemisah baris" ( baris baru dalam kasus CSV, tetapi sekali lagi itu tergantung pada format yang dibaca; tidak berfungsi jika formatnya cukup rumit).

Potongan-potongan ini, masing-masing mungkin berisi beberapa entri, kemudian dapat bertani ke kerumunan proses paralel membaca pekerjaan dari antrian, di mana mereka diurai dan dibagi, kemudian ditempatkan di antrian untuk tahap 2.


0
2018-03-10 16:39