mysql - 使用 sqlalchemy 检查行和返回 id 的存在很慢

Question

全部，

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

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

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() 来自 here ，尽管由于其他地方的限制，它已被转换为 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 很棒，但要意识到它与批量插入并不完全吻合。批量插入需要在 session 中使用较低级别的 execute 语句。这些不将 ORM 对象作为输入，而是一个字典列表和一个 insert 对象。因此，如果您将充满行的 csv 文件转换为 ORM 对象，则需要NOT将它们添加到当前 session 中，而是将它们转换为字典以备后用。

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方法来自here .这使您可以创建 ORM 对象中列的字典。它们永远不会被添加到 session 中，并在不久之后从内存中消失。

1. 人际关系会咬你一口

如果你已经建立了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=""))

确保 cascade 在 backref 中为空!否则，将关系中的对象插入到 session 中将 cascade 通过其余的对象。当您稍后尝试批量插入您的值时，它们将作为重复项被拒绝......如果您幸运的话。

这很重要，因为部分要求是获取有效地址的 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 首先检查 session ，防止跨网络传输。但是，它只有在 新 地址被添加到 session 中时才有效!还记得关系吗？这是级联到语句的插入中。此外，如果您没有 flush() 或有 autoflush=True，则 get 无法看到新添加的对象。

1. 创建 session 时，保留您的对象!

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

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

1. ORM 对象确实有插入

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

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

Buried in the docs ，似乎可以使用 classname.__table__ 作为必要的表对象，insert 要求.因此，在 session 中，使用 ORM 类获取表以获取插入对象，使用字典列表运行执行。不要忘记事后提交!

1. 不要耗尽内存

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