Masseneinsatz mit SQLAlchemy ORM

Gibt es eine Möglichkeit, SQLAlchemy dazu zu bringen, eine Masseneinfügung durchzuführen, anstatt jedes einzelne Objekt einzufügen? dh

tun:

INSERT INTO `foo` (`bar`) VALUES (1), (2), (3)

eher, als:

INSERT INTO `foo` (`bar`) VALUES (1)
INSERT INTO `foo` (`bar`) VALUES (2)
INSERT INTO `foo` (`bar`) VALUES (3)

Ich habe gerade Code konvertiert, um SQLalchemie anstelle von Roh-SQL zu verwenden, und obwohl es jetzt viel schöner ist, damit zu arbeiten, scheint es jetzt langsamer zu sein (bis zu einem Faktor von 10), frage ich mich, ob dies der Grund ist.

Vielleicht könnte ich die Situation durch effizientere Sitzungen verbessern. Im Moment habe autoCommit=Falseund mache session.commit()ich ein, nachdem ich ein paar Sachen hinzugefügt habe. Obwohl dies dazu zu führen scheint, dass die Daten veraltet sind, wenn die Datenbank an einer anderen Stelle geändert wird, wie auch wenn ich eine neue Abfrage durchführe, erhalte ich immer noch alte Ergebnisse zurück?

Danke für Ihre Hilfe!

SQLAlchemy hat das in der Version eingeführt 1.0.0:

Massenoperationen - SQLAlchemy-Dokumente

Mit diesen Vorgängen können Sie jetzt Masseneinfügungen oder Aktualisierungen vornehmen!

Zum Beispiel können Sie Folgendes tun:

s = Session()
objects = [
    User(name="u1"),
    User(name="u2"),
    User(name="u3")
]
s.bulk_save_objects(objects)
s.commit()

Hier wird ein Bulk-Einsatz gemacht.

In den sqlalchemy-Dokumenten wird die Leistung verschiedener Techniken beschrieben, die für Masseneinfügungen verwendet werden können:

ORMs sind grundsätzlich nicht für Hochleistungs-Bulk-Inserts gedacht - dies ist der ganze Grund, warum SQLAlchemy den Core zusätzlich zum ORM als erstklassige Komponente anbietet.

Für den Anwendungsfall schneller Masseneinfügungen ist das SQL-Generierungs- und Ausführungssystem, auf dem der ORM aufbaut, Teil des Kerns. Wenn Sie dieses System direkt verwenden, können Sie ein INSERT erstellen, das mit der direkten Verwendung der Rohdatenbank-API konkurrenzfähig ist.

Alternativ bietet das SQLAlchemy ORM die Bulk Operations-Suite von Methoden an, die Hooks in Unterabschnitte des Arbeitsprozesses bereitstellen, um INSERT- und UPDATE-Konstrukte auf Kernebene mit einem geringen Grad an ORM-basierter Automatisierung auszugeben.

Das folgende Beispiel zeigt zeitbasierte Tests für verschiedene Methoden zum Einfügen von Zeilen, von den am meisten automatisierten bis zu den am wenigsten automatisierten. Mit cPython 2.7 wurden Laufzeiten beobachtet:

classics-MacBook-Pro:sqlalchemy classic$ python test.py
SQLAlchemy ORM: Total time for 100000 records 12.0471920967 secs
SQLAlchemy ORM pk given: Total time for 100000 records 7.06283402443 secs
SQLAlchemy ORM bulk_save_objects(): Total time for 100000 records 0.856323003769 secs
SQLAlchemy Core: Total time for 100000 records 0.485800027847 secs
sqlite3: Total time for 100000 records 0.487842082977 sec

Skript:

import time
import sqlite3

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String,  create_engine
from sqlalchemy.orm import scoped_session, sessionmaker

Base = declarative_base()
DBSession = scoped_session(sessionmaker())
engine = None


class Customer(Base):
    __tablename__ = "customer"
    id = Column(Integer, primary_key=True)
    name = Column(String(255))


def init_sqlalchemy(dbname='sqlite:///sqlalchemy.db'):
    global engine
    engine = create_engine(dbname, echo=False)
    DBSession.remove()
    DBSession.configure(bind=engine, autoflush=False, expire_on_commit=False)
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)


def test_sqlalchemy_orm(n=100000):
    init_sqlalchemy()
    t0 = time.time()
    for i in xrange(n):
        customer = Customer()
        customer.name = 'NAME ' + str(i)
        DBSession.add(customer)
        if i % 1000 == 0:
            DBSession.flush()
    DBSession.commit()
    print(
        "SQLAlchemy ORM: Total time for " + str(n) +
        " records " + str(time.time() - t0) + " secs")


def test_sqlalchemy_orm_pk_given(n=100000):
    init_sqlalchemy()
    t0 = time.time()
    for i in xrange(n):
        customer = Customer(id=i+1, name="NAME " + str(i))
        DBSession.add(customer)
        if i % 1000 == 0:
            DBSession.flush()
    DBSession.commit()
    print(
        "SQLAlchemy ORM pk given: Total time for " + str(n) +
        " records " + str(time.time() - t0) + " secs")


def test_sqlalchemy_orm_bulk_insert(n=100000):
    init_sqlalchemy()
    t0 = time.time()
    n1 = n
    while n1 > 0:
        n1 = n1 - 10000
        DBSession.bulk_insert_mappings(
            Customer,
            [
                dict(name="NAME " + str(i))
                for i in xrange(min(10000, n1))
            ]
        )
    DBSession.commit()
    print(
        "SQLAlchemy ORM bulk_save_objects(): Total time for " + str(n) +
        " records " + str(time.time() - t0) + " secs")


def test_sqlalchemy_core(n=100000):
    init_sqlalchemy()
    t0 = time.time()
    engine.execute(
        Customer.__table__.insert(),
        [{"name": 'NAME ' + str(i)} for i in xrange(n)]
    )
    print(
        "SQLAlchemy Core: Total time for " + str(n) +
        " records " + str(time.time() - t0) + " secs")


def init_sqlite3(dbname):
    conn = sqlite3.connect(dbname)
    c = conn.cursor()
    c.execute("DROP TABLE IF EXISTS customer")
    c.execute(
        "CREATE TABLE customer (id INTEGER NOT NULL, "
        "name VARCHAR(255), PRIMARY KEY(id))")
    conn.commit()
    return conn


def test_sqlite3(n=100000, dbname='sqlite3.db'):
    conn = init_sqlite3(dbname)
    c = conn.cursor()
    t0 = time.time()
    for i in xrange(n):
        row = ('NAME ' + str(i),)
        c.execute("INSERT INTO customer (name) VALUES (?)", row)
    conn.commit()
    print(
        "sqlite3: Total time for " + str(n) +
        " records " + str(time.time() - t0) + " sec")

if __name__ == '__main__':
    test_sqlalchemy_orm(100000)
    test_sqlalchemy_orm_pk_given(100000)
    test_sqlalchemy_orm_bulk_insert(100000)
    test_sqlalchemy_core(100000)
    test_sqlite3(100000)

Soweit ich weiß, gibt es keine Möglichkeit, das ORM dazu zu bringen, Masseneinsätze auszustellen. Ich glaube, der Grund dafür ist, dass SQLAlchemy die Identität jedes Objekts (dh neue Primärschlüssel) verfolgen muss, und Bulk-Einfügungen stören dies. Angenommen, Ihre fooTabelle enthält eine idSpalte und ist einer FooKlasse zugeordnet:

x = Foo(bar=1)
print x.id
# None
session.add(x)
session.flush()
# BEGIN
# INSERT INTO foo (bar) VALUES(1)
# COMMIT
print x.id
# 1

Da SQLAlchemy den Wert für x.idohne weitere Abfrage ermittelt hat, können wir daraus schließen, dass der Wert direkt aus der INSERTAnweisung stammt. Wenn Sie keinen späteren Zugriff auf die erstellten Objekte über dieselben Instanzen benötigen , können Sie die ORM-Ebene für Ihre Einfügung überspringen:

Foo.__table__.insert().execute([{'bar': 1}, {'bar': 2}, {'bar': 3}])
# INSERT INTO foo (bar) VALUES ((1,), (2,), (3,))

SQLAlchemy kann diese neuen Zeilen nicht mit vorhandenen Objekten abgleichen, daher müssen Sie sie für nachfolgende Vorgänge erneut abfragen.

In Bezug auf veraltete Daten ist es hilfreich, sich daran zu erinnern, dass die Sitzung keine integrierte Methode enthält, um festzustellen, wann die Datenbank außerhalb der Sitzung geändert wird. Um über vorhandene Instanzen auf extern geänderte Daten zugreifen zu können, müssen die Instanzen als abgelaufen markiert werden . Dies geschieht standardmäßig am session.commit(), kann jedoch manuell durch Aufrufen von session.expire_all()oder erfolgen session.expire(instance). Ein Beispiel (SQL weggelassen):

x = Foo(bar=1)
session.add(x)
session.commit()
print x.bar
# 1
foo.update().execute(bar=42)
print x.bar
# 1
session.expire(x)
print x.bar
# 42

session.commit()läuft ab x, sodass die erste print-Anweisung implizit eine neue Transaktion öffnet und erneut abfragtx die Attribute erneut ab. Wenn Sie die erste Druckanweisung auskommentieren, werden Sie feststellen, dass die zweite jetzt den richtigen Wert aufnimmt, da die neue Abfrage erst nach der Aktualisierung ausgegeben wird.

Dies ist unter dem Gesichtspunkt der Transaktionsisolation sinnvoll - Sie sollten nur externe Änderungen zwischen Transaktionen vornehmen. Wenn dies zu Problemen führt, würde ich vorschlagen, die Transaktionsgrenzen Ihrer Anwendung zu klären oder zu überdenken, anstatt sofort danach zu greifen session.expire_all().

Ich mache es normalerweise mit add_all.

from app import session
from models import User

objects = [User(name="u1"), User(name="u2"), User(name="u3")]
session.add_all(objects)
session.commit()

SQLAlchemy wurde ab Version 0.8 direkt unterstützt

Richtet sich nach den docs , connection.execute(table.insert().values(data))sollte es tun. (Beachten Sie, dass dies nicht dasselbe ist, connection.execute(table.insert(), data)was zu vielen einzelnen Zeileneinfügungen über einen Aufruf von führt. executemany) Bei einer anderen als einer lokalen Verbindung kann der Leistungsunterschied enorm sein.

SQLAlchemy hat das in der Version eingeführt 1.0.0:

Massenoperationen - SQLAlchemy-Dokumente

Mit diesen Vorgängen können Sie jetzt Masseneinfügungen oder Aktualisierungen vornehmen!

Zum Beispiel (wenn Sie den geringsten Overhead für einfache Tabellen-INSERTs wünschen) können Sie Folgendes verwenden Session.bulk_insert_mappings():

loadme = [(1, 'a'),
          (2, 'b'),
          (3, 'c')]
dicts = [dict(bar=t[0], fly=t[1]) for t in loadme]

s = Session()
s.bulk_insert_mappings(Foo, dicts)
s.commit()

Oder, wenn Sie möchten, überspringen Sie die loadmeTupel und schreiben Sie die Wörterbücher direkt in dicts(aber ich finde es einfacher, die gesamte Worthaftigkeit aus den Daten herauszulassen und eine Liste der Wörterbücher in einer Schleife zu laden).

Die Antwort von Piere ist richtig, aber ein Problem ist, dass bulk_save_objectsstandardmäßig nicht die Primärschlüssel der Objekte zurückgegeben werden, wenn dies für Sie von Belang ist. Set return_defaultszuTrue um dieses Verhalten zu erhalten.

Die Dokumentation finden Sie hier .

foos = [Foo(bar='a',), Foo(bar='b'), Foo(bar='c')]
session.bulk_save_objects(foos, return_defaults=True)
for foo in foos:
    assert foo.id is not None
session.commit()

Alle Wege führen nach Rom , aber einige von ihnen überqueren Berge, erfordern Fähren, aber wenn Sie schnell dorthin wollen, nehmen Sie einfach die Autobahn.

In diesem Fall muss die Autobahn die Funktion execute_batch () von psycopg2 verwenden . Die Dokumentation sagt es am besten:

Die derzeitige Implementierung von executemany()ist (unter Verwendung einer äußerst gemeinnützigen Untertreibung) nicht besonders leistungsfähig. Diese Funktionen können verwendet werden, um die wiederholte Ausführung einer Anweisung anhand einer Reihe von Parametern zu beschleunigen. Durch die Reduzierung der Anzahl der Server-Roundtrips kann die Leistung um Größenordnungen besser sein als bei Verwendung executemany().

In meinem eigenen Test execute_batch()ist ungefähr doppelt so schnell wie executemany()und bietet die Möglichkeit, die page_size für weitere Optimierungen zu konfigurieren (wenn Sie die letzten 2-3% der Leistung aus dem Treiber herausholen möchten).

Dieselbe Funktion kann problemlos aktiviert werden, wenn Sie SQLAlchemy verwenden, indem use_batch_mode=TrueSie beim Instanziieren der Engine als Parameter festlegencreate_engine()

Dies ist ein Weg:

values = [1, 2, 3]
Foo.__table__.insert().execute([{'bar': x} for x in values])

Dies wird wie folgt eingefügt:

INSERT INTO `foo` (`bar`) VALUES (1), (2), (3)

Referenz: Die SQLAlchemy FAQ enthält Benchmarks für verschiedene Methoden begehen.

Die beste Antwort, die ich bisher gefunden habe, war in der sqlalchemy-Dokumentation:

http://docs.sqlalchemy.org/en/latest/faq/performance.html#im-inserting-400-000-rows-with-the-orm-and-it-s-really-slow

Es gibt ein vollständiges Beispiel für einen Benchmark möglicher Lösungen.

Wie in der Dokumentation gezeigt:

mass_save_objects ist nicht die beste Lösung, aber die Leistung ist korrekt.

Die zweitbeste Implementierung in Bezug auf die Lesbarkeit war meiner Meinung nach mit dem SQLAlchemy Core:

def test_sqlalchemy_core(n=100000):
    init_sqlalchemy()
    t0 = time.time()
    engine.execute(
        Customer.__table__.insert(),
            [{"name": 'NAME ' + str(i)} for i in xrange(n)]
    )

Der Kontext dieser Funktion ist im Dokumentationsartikel angegeben.