如何在sqlalchemy中使用psycopg2.extras？

Question

我想将大量条目（〜600k）上传到PostgreSQL DB的一个简单表中，每个条目有一个外键，一个时间戳和3个浮点数。但是，每个条目要花费60毫秒才能执行此处所述的核心批量插入操作，因此整个执行过程将花费10个小时。我发现，这是executemany()方法的性能问题，但是已经用psycopg2 2.7中的execute_values()方法解决了。

我运行的代码如下：

#build a huge list of dicts, one dict for each entry
engine.execute(SimpleTable.__table__.insert(),
               values) # around 600k dicts in a list

我看到这是一个普遍的问题，但是我还没有设法在sqlalchemy本身中找到解决方案。有什么方法可以告诉sqlalchemy execute_values()在某些情况下调用吗？还有其他方法可以实现巨大的插入而无需自己构造SQL语句吗？

谢谢您的帮助！

Answer 1

同时（从SqlAlchemy 1.2.0版开始）使用功能use_batch_mode上的标志成为可能create_engine()。请参阅文档。它使用中的execute_batch()功能psycopg.extras。

Answer 2

不是您正在寻找的答案，因为这并没有解决尝试指示 SQLAlchemy 使用 psycopg extras 的问题，并且需要 \xe2\x80\x93 排序 \xe2\x80\x93 手动 SQL，但是：您可以访问来自引擎的底层 psycopg 连接raw_connection()，允许使用COPY FROM：

\n\n

import io\nimport csv\nfrom psycopg2 import sql\n\ndef bulk_copy(engine, table, values):\n    csv_file = io.StringIO()\n    headers = list(values[0].keys())\n    writer = csv.DictWriter(csv_file, headers)\n    writer.writerows(values)\n\n    csv_file.seek(0)\n\n    # NOTE: `format()` here is *not* `str.format()`, but\n    # `SQL.format()`. Never use plain string formatting.\n    copy_stmt = sql.SQL("COPY {} (" +\n                        ",".join(["{}"] * len(headers)) +\n                        ") FROM STDIN CSV").\\\n        format(sql.Identifier(str(table.name)),\n               *(sql.Identifier(col) for col in headers))\n\n    # Fetch a raw psycopg connection from the SQLAlchemy engine\n    conn = engine.raw_connection()\n    try:\n        with conn.cursor() as cur:\n            cur.copy_expert(copy_stmt, csv_file)\n\n        conn.commit()\n\n    except:\n        conn.rollback()\n        raise\n\n    finally:\n        conn.close()\n

\n\n

进而

\n\n

bulk_copy(engine, SimpleTable.__table__, values)\n

\n\n

与执行 INSERT 语句相比，这应该快得多。在这台机器上移动 600,000 条记录大约需要 8 秒，约 13\xc2\xb5s/记录。您还可以将原始连接和光标与 extras 包一起使用。

\n