通过pymysql用多个进程更新一张MYSQL表

Posted 2023-03-23

【中文标题】通过pymysql用多个进程更新一张MYSQL表

【英文标题】：Updating one table of MYSQL with multiple processes via pymysql

【发布时间】：2019-05-25 23:32:56

【问题描述】：

实际上，我正在尝试通过pymysql 更新一个包含多个进程的表，每个进程读取一个从一个巨大的文件中拆分出来的 CSV 文件以提高速度。但是当我运行脚本时，我得到了Lock wait timeout exceeded; try restarting transaction exception。在搜索了这个网站的帖子后，我发现有一篇帖子提到了设置或构建built-in LOAD_DATA_INFILE，但没有详细说明。如何使用 'pymysql' 来达到我的目标？

---------------第一次编辑------ ----------------------

工作方法如下：

`def importprogram(path, name):
    begin = time.time()
    print('begin to import program' + name + ' info.')
    # "c:\\sometest.csv"
    file = open(path, mode='rb')
    csvfile = csv.reader(codecs.iterdecode(file, 'utf-8'))

    connection = None
    try:
        connection = pymysql.connect(host='a host', user='someuser', password='somepsd', db='mydb',
                                 cursorclass=pymysql.cursors.DictCursor)
        count = 1
        with connection.cursor() as cursor:
            sql = '''update sometable set Acolumn='guid' where someid='pid';'''
            next(csvfile, None)
            for line in csvfile:
                try:
                    count = count + 1
                    if ''.join(line).strip():
                        command = sql.format(guid=line[2], pid=line[1])
                        cursor.execute(command)
                    if count % 1000 == 0:
                        print('program' + name + ' cursor execute', count)
                except csv.Error:
                    print('program csv.Error:', count)
                    continue
                except IndexError:
                    print('program IndexError:', count)
                    continue
                except StopIteration:
                    break
    except Exception as e:
        print('program' + name, str(e))
    finally:
        connection.commit()
        connection.close()
        file.close()
    print('program' + name + ' info done.time cost:', time.time()-begin)`

以及多处理方法：

import multiprocessing as mp
def multiproccess():
    pool = mp.Pool(3)
    results = []
    paths = ['C:\\testfile01.csv', 'C:\\testfile02.csv', 'C:\\testfile03.csv']
    name = 1
    for path in paths:
        results.append(pool.apply_async(importprogram, args=(path, str(name))))
        name = name + 1

    print(result.get() for result in results)
    pool.close()
    pool.join()

以及主要方法：

if __name__ == '__main__':
    multiproccess()

我是 Python 新手。我怎样才能使代码或方式本身出错？我应该只使用一个进程来完成数据读取和导入吗？

【参考方案1】：

您的问题是您超出了从服务器获取响应所允许的时间，因此客户端会自动超时。 根据我的经验，将等待超时调整为 6000 秒，合并为一个 CSV，然后将数据留给导入。另外，我建议直接从 MySQL 而不是 Python 运行查询。

我通常将 CSV 数据从 Python 导入 MySQL 的方式是通过 INSERT ... VALUES ... 方法，并且我只在需要对数据进行某种操作时才这样做（即将不同的行插入到不同的表中） )。

我喜欢你的方法并理解你的想法，但实际上没有必要。 INSERT ... VALUES ... 方法的好处是您不会遇到任何超时问题。