带有生成器的 SQLAlchemy executemany答案

【问题标题】：SQLAlchemy exectutemany with generator带有生成器的 SQLAlchemy executemany
【发布时间】：2015-01-21 18:16:48
【问题描述】：

问题：对于连接对象的执行函数，是否可以使用返回字典而不是字典列表的生成器来执行“executemany”插入？

详细信息：我正在尝试通过core expressions 学习 SQLAlchemy。作为测试，我有一个相当大的数据集，通过迭代器从文件访问，我试图将其传输到 PostgreSQL 表中，但插入单个行非常慢（参见下面的示例 1）。根据documentation，如果传入的是字典列表而不是单个字典，则连接对象的execute() 函数将等效于executemany()。我做了一些快速测试，确实这种方法对于插入组来说要快得多。不幸的是，由于我的数据集很大，我无法在内存中创建完整的字典列表，因此我的问题...

示例 1：以下（伪）代码对于大量数据非常慢

from sqlalchemy import MetaData, Table, Column

metadata = MetaData()
data = Table('data', metadata, Column...)

engine = sql.create_engine('postgresql://user:pass$@localhost/testdb')
metadata.create_all(engine)

conn = engine.connect()
ins = data.insert()
for datum in large_data_iterator:
    datum_dict = do_some_proc(datum)
    conn.execute(ins, datum_dict)

因为 execute 可以采用多个值，所以最好将最终的 for 循环替换为以下生成器版本：

def datagen(iterator):
    for datum in large_data_iterator:
        datum_dict = do_some_proc(datum)
        yield datum_dict

conn = engine.connect()
ins = data.insert()
conn.execute(ins, datagen(large_data_iterator))

但是，这会引发以下异常：AttributeError: 'list' object has no attribute 'keys'。

有谁知道是否可以让生成器版本正常工作？或者更好的方法来做到这一点也很棒。谢谢！

注意：我测试了一个修改后的生成器表达式，该表达式生成块作为字典列表（如下），它比单个执行更快。但是，我不知道如何选择最佳的块数，而且我担心生成器代码增加的复杂性使其更容易出错。（但如果这是唯一的方法……）

def datagen(iterator):
    output = []
    N = 0
    for datum in large_data_iterator:
        datum_dict = do_some_proc(datum)
        output.append(datum_dict)
        N += 1
        if N == 100: # or whatever
            yield output
            N = 0
            output = []
    if output != []:
        yield output

【问题讨论】：

标签： python sqlalchemy

【解决方案1】：

Connection 有 execution_options，它采用 stream_results 参数，但不幸的是，在底部它说“该标志目前只能被 psycopg2 方言理解”，即使有其他驱动程序使用流式支持（例如 oursql）。

在 sqlalchemy 完全支持它之前，您可以轻松地向break any iterable into chunks 编写一个辅助函数，以避免修改生成器时容易出错。

【讨论】：