Celui qui divise le fichier csv, le lit et le traite en parallèle

Isono ~! J'ai un fichier csv qui contient des millions d'enregistrements, alors divisons-le et traitons-le en parallèle!

Remarques

――Je ne vous ai pas vu faire du fractionnement et de la mise en parallèle, alors je l'ai écrit sous forme de mémorandum, dites-moi s'il y a un bel article --python3.7 --Je n'utilise pas de pandas

• pool Utilisation
• future Non utilisé

Je vais le faire tout de suite

Fichier à lire

sample.csv

1,Ah
2,je
3,U
4,e
5,Oh

J'ai emprunté gen_chunks () parce qu'il y avait une personne qui l'a fait https://stackoverflow.com/a/4957046 Quelqu'un a déjà fait la majeure partie du travail. Merci internet, merci ancêtres Cliquez ici pour le code complété

pool.py

import csv
import time
from multiprocessing import Pool

def read():
    f = open("sample.csv", "r")
    reader = csv.reader(f)

    pool = Pool()
    results = []
    for data_list in gen_chunks(reader):
        results.append(pool.apply_async(do_something, [data_list]))
    pool.close()
    pool.join()
    _ = [r.get() for r in results]

    f.close()

def do_something(data_list):
    print(f"start {data_list}")
    time.sleep(len(data_list))
    # hoge
    print(f"finish {data_list}")

def gen_chunks(reader, chunksize=2):
    """
    Chunk generator. Take a CSV `reader` and yield
    `chunksize` sized slices.
    """
    chunk = []
    for i, line in enumerate(reader):
        if i % chunksize == 0 and i > 0:
            yield chunk
            chunk = []
        chunk.append(line)
    yield chunk

read()

résultat

start [['1', 'Ah'], ['2', 'je']]
start [['3', 'U'], ['4', 'e']]
start [['5', 'Oh']]
finish [['5', 'Oh']]
finish [['1', 'Ah'], ['2', 'je']]
finish [['3', 'U'], ['4', 'e']]

Puisque chunksize = 2, deux enregistrements sont passés à do_something (). Si c'est un gros csv, faisons le bien ici. Je récupère une erreur avec _ = [r.get () for r in results]. Je devrais gérer l'erreur, mais je l'omets car c'est gênant Il semble y avoir une meilleure façon de l'écrire, alors faites-le moi savoir si vous savez

De plus, comme indiqué dans stackoverflow, si la partie de réinitialisation du tableau de gen_chunks () est définie sur del chunk [:], la sortie sera la suivante.

résultat

start [['5', 'Oh']]
start [['5', 'Oh']]
start [['5', 'Oh']]
finish [['5', 'Oh']]
finish [['5', 'Oh']]
finish [['5', 'Oh']]

Je n'ai pas bien lu les commentaires donc j'ai été témoin d'un résultat décevant C'est triste