mysql - 断开连接后调用pymysqlexecutemany时超出锁等待超时

Question

我有一个相当大的数据集，大约有 6,000,000 行 X 60 列，我正在尝试将其插入数据库中。我将它们分块，并使用我编写的类和 pymysql 将它们一次插入 10,000 个到 mysql 数据库中。问题是，我在写入时偶尔会导致服务器超时，因此我修改了我的executemany 调用以在出现错误时重新连接。当我失去连接一次时，这工作得很好，但如果我第二次失去错误，我会得到一个 pymysql.InternalException ，指出超出了锁定等待超时。我想知道如何修改以下代码以捕获该问题并在再次尝试之前完全销毁事务。

我尝试在连接上调用 rollback() ，但是如果连接因不再有游标而被破坏，这会导致另一个 InternalException 。

任何帮助将不胜感激(我也不明白为什么我一开始就会超时，但数据相对较大。)

class Database:
    def __init__(self, **creds):
        self.conn = None
        self.user = creds['user']
        self.password = creds['password']
        self.host = creds['host']
        self.port = creds['port']
        self.database = creds['database']

    def connect(self, type=None):
        self.conn = pymysql.connect(
            host = self.host,
            user = self.user,
            password = self.password,
            port = self.port,
            database = self.database
        )

    def executemany(self, sql, data):
        while True:
            try:
                with self.conn.cursor() as cursor:
                    cursor.executemany(sql, data)
                    self.conn.commit()
                    break

            except pymysql.err.OperationalError:
                print('Connection error.  Reconnecting to database.')
                time.sleep(2)
                self.connect()
                continue

        return cursor

我这样调用它:

for index, chunk in enumerate(dataframe_chunker(df), start=1):
     print(f"Writing chunk\t{index}\t{timer():.2f}")
     db.executemany(insert_query, chunk.values.tolist())

Answer 1

看看MySQL在做什么。锁等待超时是因为在其他事情完成之前无法完成插入，这可能是您自己的代码。

SELECT * FROM `information_schema`.`innodb_locks`;

将显示当前的锁。

select * from information_schema.innodb_trx where trx_id = [lock_trx_id];

将显示涉及的交易

SELECT * FROM INFORMATION_SCHEMA.PROCESSLIST where id = [trx_mysql_thread_id];

将显示所涉及的连接，并可能显示其锁定导致锁定等待超时的查询。也许有未提交的事务。

这可能是您自己的代码，因为与您的 executemany 函数的交互会捕获异常并重新连接到数据库。之前的连接呢？锁等待超时是否会终止先前的连接？虽然这是真的，但会带来麻烦。

对于在数据库连接上调用 executemany 的代码，请对 try/except 采取更多防御措施，例如:

    def executemany(self, sql, data):
        while True:
            try:
                with self.conn.cursor() as cursor:
                    cursor.executemany(sql, data)
                    self.conn.commit()
                    break
            except pymysql.err.OperationalError:
                print('Connection error.  Reconnecting to database.')
                if self.conn.is_connected():
                    connection.close()
            finally:
                time.sleep(2)
                self.connect()

但这里的解决方案是，如果没有其他数据库客户端，则不引发锁等待超时。