MySQL中的Sqlalchemy批量更新工作非常慢

Question

我正在使用SQLAlchemy 1.0.0，并且希望批量地进行一些UPDATE ONLY查询(如果匹配，主键其他什么都不做)。

我做了一些实验，发现大容量更新看起来比大容量插入或批量upsert慢得多。

请你帮我指出为什么它的工作速度这么慢，或者有没有其他的方法/想法来制造BULK UPDATE (not BULK UPSERT) with SQLAlchemy？

下面是MYSQL中的表：

CREATE TABLE `test` (
  `id` int(11) unsigned NOT NULL,
  `value` int(11) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

测试代码：

from sqlalchemy import create_engine, text
import time

driver = 'mysql'
host = 'host'
user = 'user'
password = 'password'
database = 'database'
url = "{}://{}:{}@{}/{}?charset=utf8".format(driver, user, password, host, database)

engine = create_engine(url)
engine.connect()

engine.execute('TRUNCATE TABLE test')

num_of_rows = 1000

rows = []
for i in xrange(0, num_of_rows):
    rows.append({'id': i, 'value': i})

print '--------- test insert --------------'
sql = '''
    INSERT INTO test (id, value)
    VALUES (:id, :value)
'''
start = time.time()
engine.execute(text(sql), rows)
end = time.time()
print 'Cost {} seconds'.format(end - start)

print '--------- test upsert --------------'
for r in rows:
    r['value'] = r['id'] + 1

sql = '''
    INSERT INTO test (id, value)
    VALUES (:id, :value)
    ON DUPLICATE KEY UPDATE value = VALUES(value)
'''
start = time.time()
engine.execute(text(sql), rows)
end = time.time()
print 'Cost {} seconds'.format(end - start)

print '--------- test update --------------'
for r in rows:
    r['value'] = r['id'] * 10

sql = '''
    UPDATE test
    SET value = :value
    WHERE id = :id
'''
start = time.time()
engine.execute(text(sql), rows)
end = time.time()
print 'Cost {} seconds'.format(end - start)

当num_of_rows =100时输出：

--------- test insert --------------
Cost 0.568960905075 seconds
--------- test upsert --------------
Cost 0.569655895233 seconds
--------- test update --------------
Cost 20.0891299248 seconds

当num_of_rows =1000时输出：

--------- test insert --------------
Cost 0.807548999786 seconds
--------- test upsert --------------
Cost 0.584554195404 seconds
--------- test update --------------
Cost 206.199367046 seconds

数据库服务器的网络延迟约为500 is。

看起来像在批量更新，它发送和执行每个查询一个一个，而不是成批？

提前谢谢。

Answer 1

您可以使用技巧来加快批量更新操作，即使数据库服务器(就像在您的例子中)有一个非常糟糕的延迟。与其直接更新表，不如使用舞台表非常快地插入新数据，然后对目标表执行一次联接更新。这也有一个优点，那就是可以大大减少必须发送到数据库的语句的数量。

更新是如何工作的？

假设您有一个表entries，并且总是有新的数据，但是您只想更新那些已经存储的数据。创建目标表entries_stage的副本，其中仅包含相关字段：

entries = Table('entries', metadata,
    Column('id', Integer, autoincrement=True, primary_key=True),
    Column('value', Unicode(64), nullable=False),
)

entries_stage = Table('entries_stage', metadata,
    Column('id', Integer, autoincrement=False, unique=True),
    Column('value', Unicode(64), nullable=False),
)

然后用大容量插入插入数据。如果您使用MySQL的多值插入语法(它本机不受SQLAlchemy支持，但可以构建得不太困难)，则可以进一步加快速度。

INSERT INTO enries_stage (`id`, `value`)
VALUES
(1, 'string1'), (2, 'string2'), (3, 'string3'), ...;

最后，使用来自舞台表的值更新目标表的值，如下所示：

 UPDATE entries e
 JOIN entries_stage es ON e.id = es.id
 SET e.value = es.value
 WHERE e.value != es.value;

那你就完了。

插入呢？

当然，这也可以加快插入速度。由于舞台表中已经有数据，所以只需要发出一个INSERT INTO ... SELECT语句，其中的数据还没有在目标表中。

INSERT INTO entries (id, value)
SELECT FROM entries_stage es
LEFT JOIN entries e ON e.id = es.id
HAVING e.id IS NULL;

这方面的好处是，您不必执行INSERT IGNORE、REPLACE或ON DUPLICATE KEY UPDATE，将增加您的主键，即使它们不会做任何。