Traitement parallèle Python (multitraitement et Joblib)

introduction

Lors d'une recherche avec un traitement parallèle Python, la première chose qui apparaît est multiprocessing ou Joblib est.

Il existe divers articles de commentaires sur les deux, mais Joblib est meilleur que le multitraitement.

• La fonction à mettre en parallèle peut prendre une forme autre qu'un tableau comme argument
• Lorsque vous quittez avec Ctrl + c, le processus enfant se terminera également. --Il existe une option pour afficher automatiquement la progression globale

Puisqu'il existe de nombreuses fonctions utiles telles que, j'ai l'habitude de l'utiliser. Cependant, j'ai eu l'occasion d'utiliser le multitraitement pour une raison quelconque, et quand j'ai comparé les vitesses des deux, je n'ai pas compris la raison, mais la parallélisation avec le multitraitement était plus rapide, je vais donc laisser un mémorandum.

Environnement d'exécution

• Ubuntu 16.04.1
• Intel(R) Core(TM) i7-6700K CPU @ 4.00GHz
• 8 cœurs virtuels, 4 cœurs physiques
• Python 3.5.2

Joblib

L'installation peut être effectuée avec pip.

Pour montrer brièvement comment l'utiliser,

def process(i):
	return [{'id': j, 'sum': sum(range(i*j))} for j in range(100)]

S'il y a une fonction comme

from joblib import Parallel, delayed

result = Parallel(n_jobs=-1)([delayed(process)(n) for n in range(1000)])

Vous pouvez exécuter la fonction process en parallèle en écrivant le processus de parallélisation comme ceci. Vous pouvez changer le nombre de cœurs à utiliser en changeant le contenu de n_jobs. Il sera exécuté avec le nombre maximum de cœurs pouvant être utilisé avec -1, et s'il vaut 1, ce sera la même situation qu'une exécution simple. C'est facile et bon.

Même si vous spécifiez plus que le nombre de cœurs disponibles dans n_jobs, il ne tombera pas, et il semble que le traitement sera distribué aux cœurs disponibles comme il convient.

En plus de n_jobs, vous pouvez prendre un argument détaillé, et si vous spécifiez une valeur de 0 à 10, la progression sera sortie en fonction de la fréquence spécifiée.

multiprocessing

Étant donné que le multitraitement est une bibliothèque Python standard, il peut être utilisé sans aucune installation particulière. Il existe de nombreuses fonctions, mais la manière la plus simple de les utiliser est

from multiprocessing import Pool
import multiprocessing as multi

p = Pool(multi.cpu_count())
p.map(process, list(range(1000)))
p.close()

On dirait. Spécifiez le nombre de parallèles avec Pool, et exécutez la fonction en parallèle avec map. Vous pouvez obtenir le nombre maximum de cœurs exécutables avec multi.cpu_count (). Veuillez noter que si vous ne terminez pas le Pool créé avec close (), cela consommera de la mémoire et ce sera un gros problème.

Différence d'utilisabilité

Le plus grand avantage de joblib par rapport au multitraitement est que vous pouvez prendre autre chose qu'un tableau comme argument de la fonction à paralléliser.

Par exemple

def process(i, s):
    return [{'id': i, 'str': s, 'sum': sum(range(i*j))} for j in range(100)]

En supposant qu'il existe une fonction qui prend plusieurs arguments comme Dans joblib

strs = ['a', 'b', 'c']
result = Parallel(n_jobs=job)([delayed(process)(i,s) for i,s in enumerate(strs * 1000)])

Vous pouvez le faire comme ça, mais même si vous essayez de l'exécuter avec le multitraitement de la même manière,

p = Pool(multi.cpu_count())
strs = ['a', 'b', 'c']
p.map(process, enumerate(strs * 1000))
p.close()

Je me fâche contre l'argument et j'obtiens une erreur.

TypeError: process() missing 1 required positional argument:

Dans ce cas, la fonction d'exécution

def process(v):
    return [{'id': v[0], 'str': v[1], 'sum': sum(range(v[0]*j))} for j in range(100)]

Il est nécessaire de changer pour qu'un tableau soit pris comme argument.

Comparaison de vitesse

Maintenant, comparons la vitesse du sujet principal. Pour la fonction d'exécution

def process(n):
    return sum([i*n for i in range(100000)])

Essayez une exécution parallèle pour chacun et mesurez la vitesse.

def usejoblib(job, num):
    Parallel(n_jobs=job)([delayed(process)(n) for n in range(num)])


def usemulti(job, num):
    p = Pool(multi.cpu_count() if job < 0 else job)
    p.map(process, list(range(num)))
    p.close()

Voici le résultat de la mesure 10 fois avec un nombre différent de boucles et de la moyenne de chacune (8 tâches, l'unité est la seconde)

Les deux sont beaucoup plus rapides que sans parallélisation. Il semble qu'il n'y ait pas beaucoup de différence de vitesse entre les deux (la version multitraitement est-elle légèrement plus rapide?).

Si vous essayez d'augmenter la quantité de traitement dans une boucle

def process(i):
    return [{'id': j, 'sum': sum(range(i*j))} for j in range(1000)]

Encore une fois, le multitraitement était un peu plus rapide, mais je ne savais pas pourquoi. S'il y a une telle différence, le résultat peut être inversé en fonction de l'environnement d'exécution et de la charge de traitement de la fonction, mais dans tous les cas, ce sera beaucoup plus rapide que le traitement en boucle normal, alors soyez agressif lorsque vous effectuez un traitement en boucle avec Python. Je veux l'utiliser comme cible.

Enfin, je mettrai le code exécuté cette fois.

import time
from joblib import Parallel, delayed
from multiprocessing import Pool
import multiprocessing as multi
from more_itertools import flatten
import sys
import functools


def process(i):
    return [{'id': j, 'sum': sum(range(i*j))} for j in range(1000)]


#def process(n):
#    return sum([i*n for i in range(100000)])


def usejoblib(job, num):
    result =Parallel(n_jobs=job)([delayed(process)(n) for n in range(num)])
    return result


def usemulti(job, num):
    p = Pool(multi.cpu_count() if job < 0 else job)
    result = p.map(process, range(num))
    p.close()
    return result

if __name__ == '__main__':
    argv = sys.argv
    total = 0
    n = 1
    
    for i in range(n):
        s = time.time()
        if argv[1] == 'joblib':
            result = usejoblib(int(argv[2]),int(argv[3]))
        elif argv[1] == 'multi':
            result = usemulti(int(argv[2]),int(argv[3]))
        else:
            result = [process(j) for j in range(int(argv[3]))]
        elapsed = time.time()-s
        print('time: {0} [sec]'.format(elapsed))
        total += elapsed
        
    print('--------')
    print('average: {0} [sec]'.format(total/n))

    sums = functools.reduce(lambda x, y: x + y['sum'], list(flatten(result)), 0)
    print('total: {0}'.format(sums))
#    print('total: {0}'.format(sum(result)))