使用create_引擎将SQLAlchemy转换为MSSQL

Question

我能找到的大多数用Python显示完整MSSQL连接方法的示例在几个月前已经过时，部分原因是SQLAlchemy1.3中的一些优化。我试图复制我在文档中看到的内容

我无法使用pyodbc将SQLAlchemy连接到MSSQL Server。

我有一个本地SQL server，可从以下位置的SQL server Management Studio访问：#DESKTOP-QLSOTTG\SQLEXPRESS
数据库是：TestDB
本例中的用户名为：TestUser
本例中的密码为：TestUserPass

我想运行一个将pandas数据帧导入MSSQL数据库的测试用例（cases？），以找出最快的方法。然而，这个问题的目的是围绕连通性

信贷：我从Gord那里借了一些数据帧/更新here的代码

import pandas as pd
import numpy as np
import time
from sqlalchemy import create_engine, event
from urllib.parse import quote_plus


# for pyodbc
#engine = create_engine('mssql+pyodbc://TestUser:TestUserPAss@DESKTOP-QLSOTTG\\SQLEXPRESS:1433/TestDB?driver=ODBC+Driver+17+for+SQL+Server', fast_executemany=True)
engine = create_engine("mssql+pyodbc://TestUser:TestUserPass@DESKTOP-QLSOTTG\\SQLEXPRESS:1433/TestDB?driver=ODBC+Driver+13+for+SQL+Server", fast_executemany=True)

# test data
num_rows = 10000
num_cols = 100
df = pd.DataFrame(
    [[f'row{x:04}col{y:03}' for y in range(num_cols)] for x in range(num_rows)],
    columns=[f'col{y:03}' for y in range(num_cols)]
)

t0 = time.time()
df.to_sql("sqlalchemy_test", engine, if_exists='replace', index=None)
print(f"pandas wrote {num_rows} rows in {(time.time() - t0):0.1f} seconds")

我收到的错误如下。我将假设服务器“主动拒绝连接”是因为我的连接字符串不知怎么搞砸了，但我似乎不明白原因

OperationalError: (pyodbc.OperationalError) ('08001', '[08001] [Microsoft][ODBC Driver 13 for SQL Server]TCP Provider: No connection could be made because the target machine actively refused it.\r\n (10061) (SQLDriverConnect); [08001] [Microsoft][ODBC Driver 13 for SQL Server]Login timeout expired (0); [08001] [Microsoft][ODBC Driver 13 for SQL Server]A network-related or instance-specific error has occurred while establishing a connection to SQL Server. Server is not found or not accessible. Check if instance name is correct and if SQL Server is configured to allow remote connections. For more information see SQL Server Books Online. (10061)')
(Background on this error at: http://sqlalche.me/e/13/e3q8)

可以从SQL Server Management Studio连接数据库和用户。*

关于我可能遗漏的内容有什么想法吗？

注意事项：

• 将DESKTOP-QLSOTTG\SQLEXPRESS:1433更改为DESKTOP-QLSOTTG:1433未更改错误
• 将DESKTOP-QLSOTTG\SQLEXPRESS:1433更改为localhost:1433不会更改错误
• 将DESKTOP-QLSOTTG\SQLEXPRESS:1433更改为localhost\SQLEXPRESS:1433不会更改错误

Answer 1

您需要启用TCP/IP

在开始菜单上，单击所有程序>；Microsoft SQL Server 2012>；配置工具>；SQL Server配置管理器。 单击SQL Server 2012服务

展开SQL Server 2012网络配置节点，然后选择MSSQLServer的协议（SQL实例名称）

右键单击TCP/IP，然后单击启用

在树中选择SQL Server 2012服务。 右键单击SQL Server（SQL实例名称），然后单击重新启动

Answer 2

我将用一个完整的例子来回答这个问题，因为在这个过程中我还有一些其他的问题

此示例能够：

• 使用fast\u executemany和用户指定的内存友好分块将数据快速加载到MS SQL数据库
• 在大约0.3秒内将10000条记录（25列）加载到Microsoft SQL（MSSQL）数据库
• 在大约45秒内将1000000条记录（25列）加载到Microsoft SQL（MSSQL）数据库
• 在大约9分钟内将10000000条记录（25列）加载到Microsoft SQL（MSSQL）数据库
• 使用预配置的函数分块数据，该函数可避免使用chunksize，这会在较大的数据集上导致内存错误Credit用于分块
• 可以帮助您将数据加载到ram较低的SQL server中。大容量加载占用大量ram，较小的加载大小占用的内存要少得多
• 可以添加一个try/except语句来捕获您正在尝试记录的块的加载错误/稍后重试类型设置

我已经为一些其他的DB提供者提供了一些未经测试的连接字符串。 截至2020年12月，pandas、sqlalchemy、pyodbc等的当前版本

%%time #remove this if you are not using a Jupyter notebook and just want to run a .py script

import pandas as pd
import numpy as np
import sqlalchemy as sql
import sys
import math

# Enterprise DB to be used
DRIVER = "ODBC Driver 17 for SQL Server"
USERNAME = "TestUser"
PSSWD = "TestUser"
SERVERNAME = "DESKTOP-QLSOTTG"
INSTANCENAME = "\SQLEXPRESS"
DB = "TestDB"
TABLE = "perftest"


conn_executemany = sql.create_engine(
    f"mssql+pyodbc://{USERNAME}:{PSSWD}@{SERVERNAME}{INSTANCENAME}/{DB}?driver={DRIVER}", fast_executemany=True
)



def chunker(seq, size):
    return (seq[pos : pos + size] for pos in range(0, len(seq), size))


def insert_with_progress(df, engine, table="", schema=""):
    con = engine.connect()

    # Replace table
    engine.execute(f"DROP TABLE IF EXISTS {schema}.{table};")

    # Insert with progress
    SQL_SERVER_CHUNK_LIMIT = 100000
    chunksize = math.floor(SQL_SERVER_CHUNK_LIMIT / len(df.columns))

    for chunk in chunker(df, chunksize):
        chunk.to_sql(
            name=table,
            con=con,
            if_exists="append",
            index=False
        )
        
df = pd.DataFrame(np.random.random((10 ** 7, 24)))
df['TextCol'] = "Test Goes Here"
df.head()
print("DataFrame is", round(sys.getsizeof(df) / 1024 ** 2, 1), "MB")
print("DataFrame contains", len(df), "rows by", len(df.columns), "columns")


# Doing it like this errors out. Can't seem to be able to debug the straight pandas call.
# df.to_sql(TABLE, conn_sqlalchemy, index=False, if_exists='replace', method='multi', chunksize=2100)

insert_with_progress(df, conn_executemany, table=TABLE)

关于连接字符串：

1. 如果您希望更改为另一个DB类型，您很可能只需要更改以f"mssql+pyodbc://开头的行
2. 如果SQL server不使用实例名称（例如SQLSERVERNAME\instance\u name），则可以将实例名称参数设置为空
3. 如果确实使用实例名，请确保在变量开头保留\
4. 如果使用不同的连接字符串，还需要在上面的代码窗口的最后一行用连接字符串名称替换变量名称

其他提供商的替代包含声明
这些措施包括：

• pymssql
• turbobdc

import pymssql as ms
import sqlalchemy as sql
import sqlalchemy_turbodbc as st

备用连接字符串
Credit用于DSN样式的字符串，我已将其修改为使用用户名/密码

conn_sqlalchemy = sql.create_engine(f"mssql+pyodbc://{USERNAME}:{PSSWD}@{SERVERNAME}{INSTANCENAME}/{DB}?driver={DRIVER}")

conn_executemany = sql.create_engine(
    f"mssql+pyodbc://{USERNAME}:{PSSWD}@{SERVERNAME}{INSTANCENAME}/{DB}?driver={DRIVER}", fast_executemany=True
)

conn_turbodbc = sql.create_engine(f"mssql+turbodbc://{USERNAME}:{PSSWD}@{SERVERNAME}{INSTANCENAME}/{DB}?driver={DRIVER}")

conn_pymssql = sql.create_engine(f"mssql+pymssql://{USERNAME}:{PSSWD}@{SERVERNAME}{INSTANCENAME}/{DB}")