与 pandas 和 GNU Parallel 并行将数据导入 MySQL答案

【问题标题】：Importing data to MySQL in parallel with pandas and GNU Parallel与 pandas 和 GNU Parallel 并行将数据导入 MySQL
【发布时间】：2019-01-13 19:21:16
【问题描述】：

我有几千个目录，我想从中将数据导入 MySQL。我创建了一个 python 脚本，它从单个目录读取数据并将其放入数据库。这是数据被发送到数据库的部分：

host = 'localhost'
engine = create_engine('mysql://user:pass@%s/db?charset=utf8' % host)
conn = engine.connect()
trans = conn.begin()
try:
    conn.execute('delete from tests where ml="%s"' % ml)
    tests.to_sql(con=conn, name='tests', if_exists='append', index=False)
    data.to_sql(con=conn, name='data', if_exists='append', index=False)
    trans.commit()
    print(CGRE + ml + ': OK' + CEND)
except:
    trans.rollback()
    print(CRED + ml + ': database error!' + CEND)
    raise
conn.close()

单线程执行效果不错，但是太慢了：

parallel -j 1 "[[ -d {} ]] && (cd {} && data_to_db.py) || echo {} >> ~/Data/failed_db" ::: *

现在我想启动几个进程：

parallel -j 8 .........

在执行过程中有时会出现此错误：

sqlalchemy.exc.InternalError: (pymysql.err.InternalError) (1213, '尝试获取锁时发现死锁；尝试重启事务')

有没有办法增加事务的等待时间或以其他方式解决，因为没有并行执行，导入所有数据需要很长时间？

【问题讨论】：

stackoverflow.com/questions/23697925/…
@RomanPerekhrest 谢谢，有用的链接

标签： python mysql pandas sqlalchemy gnu-parallel

【解决方案1】：

非常感谢@RomanPerekhrest，这是来自 MySQL 手册的工作 solution，使用了LOCK/UNLOCK TABLES。

engine = create_engine('mysql://user:pass@%s/db?charset=utf8' % host)
conn = engine.connect()
trans = conn.begin()
try:
    conn.execute('set autocommit=0')
    conn.execute('lock tables tests write, data write')
    conn.execute('delete from tests where ml="%s"' % ml)
    tests.to_sql(con=conn, name='tests', if_exists='append', index=False)
    data.to_sql(con=conn, name='data', if_exists='append', index=False)
    trans.commit()
    conn.execute('unlock tables')
    print(CGRE + ml + ': OK' + CEND)
except:
    trans.rollback()
    conn.execute('unlock tables')
    conn.close()
    print(CRED + ml + ': database error!' + CEND)
    raise
conn.close()

【讨论】：