SQL Chemy ORM을 사용한 벌크 인서트

SQL Chemy ORM을 사용한 벌크 인서트

SQLechemy가 각각의 오브젝트를 삽입하지 않고 대량 삽입할 수 있는 방법이 있습니까?

실행:

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

대신:

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

raw sql이 아닌 sqlalchemy를 사용하기 위해 몇 가지 코드를 변환했습니다만, 현재는 (10배까지) 느린 것 같습니다만, 이것이 원인인지 궁금합니다.

세션을 통해 상황을 좀 더 효율적으로 개선할 수 있을 것 같습니다. 나는 는나 at를 있다.autoCommit=False ★★★★★★★★★★★★★★★」session.commit()몇 가지 추가 후에요.다른 곳에서 DB를 변경하면 데이터가 오래되어 버리는 것 같습니다만, 예를 들어 새로운 쿼리를 실행해도 오래된 결과가 반환됩니까?

도와주셔서 감사합니다!

Chemy는 Chemy 버전에서 이를 했습니다.1.0.0:

대량 운용 - SQL Chemy 문서

이러한 작업을 통해 대량 삽입 또는 업데이트를 수행할 수 있습니다!

예를 들어 다음과 같은 작업을 수행할 수 있습니다.

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

여기서 대량 삽입이 이루어집니다.

sqlalchemy 문서에는 벌크 삽입에 사용할 수 있는 다양한 기술의 성능에 대한 설명이 나와 있습니다.

ORM은 기본적으로 고성능 벌크 인서트용이 아닙니다.이것이 SQL Chemy가 ORM과 더불어 Core를 1등급 컴포넌트로 제공하는 이유입니다.

고속 벌크 삽입의 경우 ORM이 구축한SQL 생성 및 실행 시스템은 Core의 일부입니다.이 시스템을 직접 사용함으로써 raw database API를 직접 사용하는 것과 경쟁력 있는 INSERT를 제작할 수 있습니다.

또는 SQLAlchemy ORM은 ORM 기반 자동화를 통해 코어 레벨의 INSERT 및 UPDATE 구조를 내보낼 수 있도록 작업 프로세스 유닛의 서브섹션에 후크를 제공하는 일괄 작업 스위트를 제공합니다.

다음 예제에서는 가장 자동화된 행에서 가장 작은 행으로 행을 삽입하는 여러 가지 다른 방법에 대한 시간 기반 테스트를 보여 줍니다.cPython 2.7에서 관찰된 런타임:

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

스크립트:

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)

제가 알기로는 ORM에 벌크 인서트를 발행하는 방법은 없는 것으로 알고 있습니다.SQL Chemy가 각 오브젝트의 ID(즉, 새로운 프라이머리 키)를 추적해야 하기 때문에 벌크 삽입이 방해되기 때문이라고 생각합니다.를 들어, '아까보다'라고 가정하면요.foo테이블에는 다음이 포함됩니다.id이 되어 .Foo 링크:

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

에서 SQL Chemy 값을 했기 때문에x.id 않고 그 을 알 수 있습니다.INSERT진술.동일한 인스턴스를 통해 생성된 개체에 대한 후속 액세스가 필요하지 않은 경우 삽입을 위해 ORM 계층을 건너뛸 수 있습니다.

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

SQL Chemy는 이러한 새 행을 기존 개체와 일치시킬 수 없으므로 이후 작업을 위해 해당 행을 다시 쿼리해야 합니다.

오래된 데이터에 관한 한 세션에는 데이터베이스가 세션 외부에서 변경되는 시기를 알 수 있는 기본 제공 방법이 없다는 점을 기억하면 도움이 됩니다.기존 인스턴스를 통해 외부에서 수정된 데이터에 액세스하려면 해당 인스턴스를 만료됨으로 표시해야 합니다.이것은 디폴트로session.commit()으로 할 수 「수동」으로 콜 해 주세요.session.expire_all() ★★★★★★★★★★★★★★★★★」session.expire(instance) 생략 (SQL 략):

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()이 만료되다x 첫 을 암묵적으로 x프린트 스테이트먼트를 후에 두 프린트 첫 번째 인쇄문을 코멘트 아웃하면 새 쿼리가 업데이트될 때까지 출력되지 않기 때문에 두 번째 인쇄문이 올바른 값을 선택한다는 것을 알 수 있습니다.

이는 트랜잭션 격리라는 관점에서 타당합니다. 트랜잭션 간에는 외부 수정 사항만 선택해야 합니다. 있는 는, 「어플리케이션의 트랜잭션 」에 경계를 명확하게 하는 것을 session.expire_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()

SQL Chemy에 대한 직접 지원은 버전 0.8에서 추가되었습니다.

의사들에 따르면connection.execute(table.insert().values(data))효과가 있을 거야(이것은, 다음과 같은 것은 아닙니다.connection.execute(table.insert(), data), 다수의 행 행 삽입됩니다.executemany로컬 접속 이외의 접속에서는 퍼포먼스의 차이가 클 수 있습니다.

Chemy는 Chemy 버전에서 이를 했습니다.1.0.0:

대량 운용 - SQL Chemy 문서

이러한 작업을 통해 대량 삽입 또는 업데이트를 수행할 수 있습니다!

예를 들어(심플한 테이블 INSERT의 오버헤드를 최소화하는 경우) 다음과 같이 사용할 수 있습니다.

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()

그냥 도 있어요.loadme 번역하다dicts(단, 모든 어휘를 데이터에서 빼서 사전 목록을 루프 형태로 로드하는 것이 더 쉽다는 것을 알게 되었습니다).

한 는 '이러다'는 것이다.bulk_save_objects문제가 있는 경우 기본적으로는 개체의 기본 키는 반환되지 않습니다. ★★return_defaults로로 합니다.True이 행동을 취하기 위해서요

설명서는 여기 있습니다.

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()

방법은 다음과 같습니다.

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

다음과 같이 삽입됩니다.

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

레퍼런스:SQL Chemy FAQ에는 다양한 커밋 방법에 대한 벤치마크가 포함되어 있습니다.

모든 길은 로마로 통하지만, 그 중 일부는 산을 넘고, 페리가 필요하지만, 빨리 가고 싶다면 고속도로를 타세요.

---

이 경우 고속도로는 psycopg2의 execute_batch() 기능을 사용합니다.문서에는 다음과 같이 기재되어 있습니다.

의 현재 executemany()(극히 자선적인 절제 표현을 사용하여) 특별히 퍼포먼스를 하고 있지 않습니다.이러한 함수를 사용하여 일련의 파라미터에 대한 스테이트먼트의 반복 실행 속도를 높일 수 있습니다.서버 라운드 트립 횟수를 줄임으로써 퍼포먼스는 사용 시보다 훨씬 향상됩니다.executemany().

의 테스트에서는execute_batch()보다 약 2배 고속입니다.executemany()및 page_size를 추가 조정을 위해 설정하는 옵션이 있습니다(드라이버에서 퍼포먼스의 마지막 2~3%를 압축하는 경우).

Chemy 를 Chemy 를 설정하면 을 쉽게 할 수 .use_batch_mode=True하여 엔진을 할 때 합니다.create_engine()

지금까지 찾은 최고의 답은 sqlalchemy 문서에서 찾을 수 있었습니다.

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

생각할 수 있는 솔루션의 벤치마크의 완전한 예를 다음에 나타냅니다.

매뉴얼에 기재되어 있는 바와 같이 다음과 같습니다.

bulk_save_module은 최적의 솔루션은 아니지만 성능은 정확합니다.

가독성 측면에서 두 번째로 뛰어난 구현은 SQL Chemy 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)]
    )

이 함수의 컨텍스트는 설명서 문서에 나와 있습니다.

Sqlalchemy는 일괄 삽입

bulk_list = [
    Foo(
        bar=1,
    ),
    Foo(
        bar=2,
    ),
    Foo(
        bar=3,
    ),
]
db.session.bulk_save_objects(bulk_list)
db.session.commit()

언급URL : https://stackoverflow.com/questions/3659142/bulk-insert-with-sqlalchemy-orm