Записывать большие базы данных Pandas в базу данных SQL Server

Question

У меня 74 относительно больших Pandas DataFrames (около 34 600 строк и 8 столбцов), которые я пытаюсь как можно быстрее вставить в базу данных SQL Server. Проведя некоторые исследования, я узнал, что хорошая функция ole pandas.to_sql не подходит для таких больших вставок в базу данных SQL Server, которая была первоначальным подходом, который я принял (очень медленно - почти час для приложения, чтобы завершить vs около 4 минут . при использовании базы данных MySQL)

This article, и многие других сообщений StackOverflow были полезны, указывая мне в правильном направлении, однако я ударил контрольно-пропускной пункт:

Я пытаюсь использовать Ядро SQLAlchemy, а не ORM по причинам, указанным в ссылке выше. Итак, я преобразование dataframe в словарь, используя pandas.to_dict и затем делать в execute() и insert():

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

Проблема заключается в том, что вставка не получает каких-либо значений - они появляются в кучу пустых скобок и I получить эту ошибку:

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

Есть значения в списке словарей, которые я прошел в, так что я не могу понять, почему значения не отображаются.

EDIT:

Вот пример, который я иду прочь:

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) + " secs") 

Answer 1

У меня есть печальная новость для вас, SQLAlchemy фактически не осуществляет массового импорта для SQL Server, это на самом деле просто собирается делать те же самые медленные индивидуальные инструкции INSERT, которые делает to_sql. Я бы сказал, что лучше всего попытаться что-то придумать, используя инструмент командной строки bcp. Вот скрипт, который я использовал в прошлом, но не гарантирует:

from subprocess import check_output, call 
import pandas as pd 
import numpy as np 
import os 

pad = 0.1 
tablename = 'sandbox.max.pybcp_test' 
overwrite=True 
raise_exception = True 
server = 'P01' 
trusted_connection= True 
username=None 
password=None 
delimiter='|' 
df = pd.read_csv('D:/inputdata.csv', encoding='latin', error_bad_lines=False) 



def get_column_def_sql(col): 
    if col.dtype == object: 
     width = col.str.len().max() * (1+pad) 
     return '[{}] varchar({})'.format(col.name, int(width)) 
    elif np.issubdtype(col.dtype, float): 
     return'[{}] float'.format(col.name) 
    elif np.issubdtype(col.dtype, int): 
     return '[{}] int'.format(col.name) 
    else: 
     if raise_exception: 
     raise NotImplementedError('data type {} not implemented'.format(col.dtype)) 
     else: 
     print('Warning: cast column {} as varchar; data type {} not implemented'.format(col, col.dtype)) 
     width = col.str.len().max() * (1+pad) 
     return '[{}] varchar({})'.format(col.name, int(width)) 

def create_table(df, tablename, server, trusted_connection, username, password, pad):   
    if trusted_connection: 
     login_string = '-E' 
    else: 
     login_string = '-U {} -P {}'.format(username, password) 

    col_defs = [] 
    for col in df: 
     col_defs += [get_column_def_sql(df[col])] 

    query_string = 'CREATE TABLE {}\n({})\nGO\nQUIT'.format(tablename, ',\n'.join(col_defs))  
    if overwrite == True: 
     query_string = "IF OBJECT_ID('{}', 'U') IS NOT NULL DROP TABLE {};".format(tablename, tablename) + query_string 


    query_file = 'c:\\pybcp_tempqueryfile.sql' 
    with open (query_file,'w') as f: 
     f.write(query_string) 

    if trusted_connection: 
     login_string = '-E' 
    else: 
     login_string = '-U {} -P {}'.format(username, password) 

    o = call('sqlcmd -S {} {} -i {}'.format(server, login_string, query_file), shell=True) 
    if o != 0: 
     raise BaseException("Failed to create table") 
    # o = call('del {}'.format(query_file), shell=True) 


def call_bcp(df, tablename): 
    if trusted_connection: 
     login_string = '-T' 
    else: 
     login_string = '-U {} -P {}'.format(username, password) 
    temp_file = 'c:\\pybcp_tempqueryfile.csv' 

    #remove the delimiter and change the encoding of the data frame to latin so sql server can read it 
    df.loc[:,df.dtypes == object] = df.loc[:,df.dtypes == object].apply(lambda col: col.str.replace(delimiter,'').str.encode('latin')) 
    df.to_csv(temp_file, index = False, sep = '|', errors='ignore') 
    o = call('bcp sandbox.max.pybcp_test2 in c:\pybcp_tempqueryfile.csv -S "localhost" -T -t^| -r\n -c')