multiprocessing.Pool pour exécuter des sous-processus simultanés pour canaliser les données à partir de mysqldump

Question

J'utilise Python pour acheminer des données d'une base de données mysql à une autre. Voici une version légèrement abstraite de code que j'utilise depuis plusieurs mois, qui a travaillé assez bien:

def copy_table(mytable): 
    raw_mysqldump = "mysqldump -h source_host -u source_user --password='secret' --lock-tables=FALSE myschema mytable" 
    raw_mysql = "mysql -h destination_host -u destination_user --password='secret' myschema" 

    mysqldump = shlex.split(raw_mysqldump) 
    mysql = shlex.split(raw_mysql) 

    ps = subprocess.Popen(mysqldump, stdout=subprocess.PIPE) 
    subprocess.check_output(mysql, stdin=ps.stdout) 
    ps.stdout.close() 
    retcode = ps.wait() 
    if retcode == 0: 
     return mytable, 1 
    else: 
     return mytable, 0 

La taille des données a augmenté, et il prend actuellement environ une heure pour copier quelque chose comme 30 les tables. Pour accélérer les choses, j'aimerais utiliser le multitraitement. J'essaye d'exécuter le code suivant sur un serveur d'Ubuntu, qui est un t2.micro (AWS EC2).

def copy_tables(tables): 
    with multiprocessing.Pool(processes=4) as pool: 
     params = [(arg, table) for table in sorted(tables)] 
     results = pool.starmap(copy_table, params) 
    failed_tables = [table for table, success in results if success == 0] 
    all_tables_processed = False if failed_tables else True 
    return all_tables_processed 

Le problème est: presque toutes les tables copiera, mais il y a toujours deux processus enfants laissés qui ne se terminera pas - ils pendent juste, et je peux voir de surveiller la base de données qu'aucune donnée est en cours de transfert. Il semble qu'ils se soient déconnectés du processus parent ou que les données ne soient pas renvoyées correctement.

Ceci est ma première question, et j'ai essayé d'être à la fois spécifique et concis - merci d'avance pour toute aide, et s'il vous plaît laissez-moi savoir si je peux fournir plus d'informations.

Answer 1

Je pense que le code suivant

ps = subprocess.Popen(mysqldump, stdout=subprocess.PIPE) 
subprocess.check_output(mysql, stdin=ps.stdout) 
ps.stdout.close() 
retcode = ps.wait() 

devrait être

ps = subprocess.Popen(mysqldump, stdout=subprocess.PIPE) 
sps = subprocess.Popen(mysql, stdin=ps.stdout) 
retcode = ps.wait() 
ps.stdout.close() 
sps.wait() 

Vous ne devriez pas fermer le tuyau jusqu'à ce que le processus de mysqldump terminé. Et check_output bloque, il va se bloquer jusqu'à ce que stdin atteigne la fin.