mysql - multiprocessing.Pool 执行并发子进程以从 mysqldump 传输数据

Question

我正在使用 Python 将数据从一个 mysql 数据库传输到另一个数据库。这是我已经使用了几个月的代码的轻度抽象版本，效果相当好:

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

数据的大小已经增加，目前复制大约 30 个表大约需要一个小时。为了加快速度，我想利用多重处理。我尝试在 Ubuntu 服务器(即 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

问题是:几乎所有的表都会复制，但总是有一些子进程无法完成 - 它们只是挂起，而且我可以通过监视数据库看到没有数据正在传输。感觉好像它们以某种方式与父进程断开了连接，或者数据没有正确返回。

这是我的第一个问题，我尽力做到具体和简洁 - 提前感谢您的帮助，如果我可以提供更多信息，请告诉我。

Answer 1

我认为下面的代码

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

应该是

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

在 mysqldump 进程完成之前，不应关闭管道。并且 check_output 是阻塞的，它将挂起直到 stdin 到达末尾。