使用 sqlalchemy 检查行是否存在并返回 id 很慢

Question

所有，

我正在读取 csv 文件并使用 sqlalchemy 将数据添加到 MySQL 数据库。其中一张表是地址表，它只应该保存唯一的地址。这些地址和另一个“语句”表之间存在关系，该表具有地址 id 的外键字段。

因此，对于数据文件中的每一行，我创建一个新语句 obj，然后获取关联地址的 id。如果地址已经存在，则返回该 ID。否则，我创建一个新地址 obj 并返回该 id。这是使用下面的代码完成的，改编自 this SO问题。

def get_or_create(self, model, rec):
    instance = self.session.query(model).filter_by(**dict(filter(lambda (x,y): x in model.__dict__.keys(), rec.iteritems()))).first()
    if instance:
        return instance
    else:
        instance = model(rec)
        return instance

我在 id 字段中使用 GUID，它是地址表主键的一部分：

class address(Base):
    __tablename__ = 'address'
    id = id_column()
    name               = Column(String(75), primary_key=True)
    Address_Line_One   = Column(String(50), primary_key=True)
    Address_Line_Two   = Column(String(50), primary_key=True)
    Address_Line_Three = Column(String(50), primary_key=True)
    Address_Line_Four  = Column(String(50), primary_key=True)

id_column() 来自 此处，尽管它已转换为CHAR(32) 由于其他地方的限制。最后，这里有一个片段：

currStatement   = statements(rec, id=currGUID)
currStatement.address = self.get_or_create(address, rec)

这一切都很好，只是速度很慢。对于在一个事务中插入约 65,000 条语句，我发现在干净的测试数据库上插入时间为 1.5 小时。实时观察插入显示它很快达到约 10,000 行，然后插入速度开始下降。

我可以做什么来加快插入时间？

编辑：

经过进一步测试，我发现插入时间缓慢的部分原因是每个对象都是单独插入的。因此，我有大约 65,000 行，每一行都成为几个 sqlalchemy 对象，单独插入。使用 sqlalchemy 0.7，如何批量插入对象？

Answer 1

好吧！

所以答案是我单独插入每一行，并为每个地址检查往返数据库。地址检查是最糟糕的部分，因为它的速度呈指数级下降。我计算出插入原始数据（1.5 小时），然后再次插入相同的数据，大约需要 9 小时！

因此，这个答案将回顾我为转换为批量插入语句所做的事情，以及一些需要注意的事情。

1. sqlalchemy 中的 ORM 将“提供帮助”

ORM 很棒，但意识到它与批量插入并不能很好地配合。批量插入需要在会话上使用较低级别的execute语句。它们不将 ORM 对象作为输入，而是将字典列表和 insert 对象作为输入。因此，如果您将充满行的 csv 文件转换为 ORM 对象，则需要不将它们添加到当前会话，而是将它们转换为字典以供以后使用。

def asdict(obj):
    return dict((col.name, getattr(obj, col.name))
         for col in class_mapper(obj.__class__).mapped_table.c)

currGUID = uuid.uuid4()
currPrintOrMail = printOrMail(rec, id=currGUID)
currStatement   = statements(rec, id=currGUID)
currAddress = self.get_or_create(address, rec)
currStatement.address = currAddress

self.currPrintOrMail_bulk.append(asdict(currPrintOrMail))
self.currStatement_bulk.append(asdict(currStatement))

asdict 方法来自此处。这将为您提供所创建的 ORM 对象中的列的字典。它们永远不会被添加到会话中，并且很快就会从内存中消失。

2. 关系会困扰你

如果你已经设置了 ORM 关系：

class statements(Base):
    __tablename__ = 'statements'
    id = id_column()
    county   = Column(String(50),default='',nullable=False)

    address_id = Column(CHAR(36), ForeignKey('address.id'))
    address = relationship("address", backref=backref("statements", cascade=""))

    printOrMail_id = Column(CHAR(36), ForeignKey('printOrMail.id'))
    pom = relationship("printOrMail", backref=backref("statements", cascade=""))

    property_id = Column(CHAR(36), ForeignKey('property.id'))
    prop = relationship("property", backref=backref("statements", cascade=""))

确保 backref 中的级联为空！否则，将关系中的对象插入会话将 级联通过其余对象。当您稍后尝试批量插入值时，它们将被视为重复项而被拒绝......如果您幸运的话。

这很重要，因为部分要求是获取有效地址的 address_id（如果存在），如果不存在则添加该地址。由于查询往返速度非常慢，我将 get_or_create 更改为：

def get_or_create(self, model, rec):
    """Check if current session has address. If not, query DB for it. If no one has the address, create and flush a new one to the session."""
    instance = self.session.query(model).get((rec['Name'], rec['Address_Line_One'], rec['Address_Line_Two'], rec['Address_Line_Three'], rec['Address_Line_Four']))
    if instance:
        return instance
    else:
        instance = model(rec)
        self.session.add(instance)
        self.session.flush()
        return instance

使用 get 导致 sqlalchemy 首先检查会话，从而防止跨网络传输。但是，只有在会话中添加新地址时它才有效！还记得关系吗？这被级联到声明的插入中。另外，如果您没有 flush() 或有 autoflush=True ，则 get 无法看到新添加的对象。

3. 创建会话时，保留您的对象！

   self.session = sessionmaker(autoflush=False, expire_on_commit=False)

如果您不包含 expire_on_commit=False 那么您将丢失地址，并再次开始往返。

4. ORM 对象确实具有插入功能

现在我们已经获得了要插入的 ORM 对象的字典列表。但我们还需要一个插入对象。

self.session.execute(printOrMail.__table__.insert(), self.currPrintOrMail_bulk)
self.session.execute(statements.__table__.insert(), self.currStatement_bulk)

埋藏在文档中，它似乎可以使用 classname.__table__ 作为 插入。因此，在会话中，使用 ORM 类获取表来获取插入对象，并使用字典列表运行执行。事后别忘了承诺！

5. 不要耗尽内存

这将允许您成功地将批量插入和 ORM 与关系以及查询 sqlalchemy 中的唯一条目混合在一起。只要注意内存不足即可。我必须一次批量插入 ~30,000 记录，否则 py2.7(32bit) 在使用大约 2G 时会崩溃。