相关疑难解决方法(0)

我有一些相当大的pandas DataFrames,我想使用新的批量SQL映射通过SQL Alchemy将它们上传到Microsoft SQL Server.pandas.to_sql方法虽然不错,但速度很慢.

我在编写代码时遇到了麻烦......

我希望能够将这个函数传递给我正在调用的pandas DataFrame,我正在调用table的模式名称schema,以及我正在调用的表名name.理想情况下,该函数将1.)删除表,如果它已经存在.2.)创建一个新表3.)创建一个mapper和4.)使用mapper和pandas数据批量插入.我被困在第3部分.

这是我的(诚然粗糙的)代码.我正在努力解决如何让mapper函数与我的主键一起工作.我真的不需要主键,但映射器功能需要它.

感谢您的见解.

from sqlalchemy import create_engine Table, Column, MetaData
from sqlalchemy.orm import mapper, create_session
from sqlalchemy.ext.declarative import declarative_base
from pandas.io.sql import SQLTable, SQLDatabase

def bulk_upload(table, schema, name):
    e = create_engine('mssql+pyodbc://MYDB')
    s = create_session(bind=e)
    m = MetaData(bind=e,reflect=True,schema=schema)
    Base = declarative_base(bind=e,metadata=m)
    t = Table(name,m)
    m.remove(t)
    t.drop(checkfirst=True)
    sqld = SQLDatabase(e, schema=schema,meta=m)
    sqlt = SQLTable(name, sqld, table).table
    sqlt.metadata = m
    m.create_all(bind=e,tables=[sqlt])    
    class MyClass(Base):
        return
    mapper(MyClass, sqlt)    

    s.bulk_insert_mappings(MyClass, table.to_dict(orient='records'))
    return

我有74个相对较大的Pandas DataFrames(大约34,600行和8列),我试图尽快插入SQL Server数据库.在做了一些研究之后,我了解到良好的ole pandas.to_sql函数对于SQL Server数据库中的大型插入是不利的,这是我采用的最初方法(非常慢 - 应用程序完成的时间差不多一小时,大约4分钟时使用mysql数据库.)

这篇文章以及许多其他StackOverflow帖子都有助于我指出正确的方向,但是我遇到了障碍:

我试图使用SQLAlchemy的Core而不是ORM,原因在上面的链接中解释.所以,我的数据帧转换为字典,使用pandas.to_dict,然后做一个execute()和insert():

self._session_factory.engine.execute(
    TimeSeriesResultValues.__table__.insert(),
    data)
# 'data' is a list of dictionaries.

问题是insert没有得到任何值 - 它们显示为一堆空括号,我得到这个错误:

(pyodbc.IntegretyError) ('23000', "[23000] [FreeTDS][SQL Server]Cannot
insert the value NULL into the column...

我传入的词典列表中有值,所以我无法弄清楚为什么值没有显示出来.

编辑:

这是我要离开的例子:

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