Python 3.6 连接到 MS SQL Server 以处理大型数据帧

Question

我是一名新的 Python 程序员，也是一名新的数据科学家，所以请原谅这里任何听起来愚蠢的事情。除非有人好奇，否则我将保留详细信息，但基本上我需要连接到 Microsoft SQL Server 并上传相对较大(约 500k 行)的 Pandas DF，并且我几乎每天都需要按照项目目前的情况执行此操作。

它不一定是 Pandas DF - 我读过有关使用 odo 处理 csv 文件的内容，但我无法让任何东西发挥作用。我遇到的问题是我无法批量插入 DF，因为该文件与 SQL Server 实例不在同一台计算机上。我不断收到如下错误:

pyodbc.ProgrammingError: ('42000', "[42000] [Microsoft][ODBC SQL Server Driver][SQL Server]Incorrect syntax near the keyword 'IF'. (156) (SQLExecDirectW)")

由于我尝试了不同的 SQL 语句，您可以将 IF 替换为 CREATE 语句中的第一个 COL_NAME 。我正在使用 SQLAlchemy 创建引擎并连接到数据库。这可能是不言而喻的，但 pd.to_sql() 方法对于我移动的数据量来说太慢了，所以这就是为什么我需要更快的方法。

顺便说一句，我正在使用 Python 3.6。我在这里列出了我尝试过但尚未成功的大部分内容。

import pandas as pd
from sqlalchemy import create_engine
import numpy as np    
df = pd.DataFrame(np.random.randint(0,100,size=(100, 1)), columns=list('test_col'))
address = 'mssql+pyodbc://uid:pw@server/path/database?driver=SQL Server'
engine = create_engine(address)
connection = engine.raw_connection()
cursor = connection.cursor()
# Attempt 1 <- This failed to even create a table at the cursor_execute statement so my issues could be way in the beginning here but I know that I have a connection to the SQL Server because I can use pd.to_sql() to create tables successfully (just incredibly slowly for my tables of interest)
create_statement = """
DROP TABLE test_table
CREATE TABLE test_table (test_col)
"""
cursor.execute(create_statement)
test_insert = '''
INSERT INTO test_table
(test_col)
values ('abs');
'''
cursor.execute(test_insert)

Attempt 2 <- From iabdb WordPress blog I came across
def chunker(seq, size):
    return (seq[pos:pos + size] for pos in range(0, len(seq), size))
records = [str(tuple(x)) for x in take_rates.values]

insert_ = """
INSERT INTO test_table
("A")
VALUES
"""

for batch in chunker(records, 2): # This would be set to 1000 in practice I hope
    print(batch)
    rows = str(batch).strip('[]')
    print(rows)
    insert_rows = insert_ + rows
    print(insert_rows)
    cursor.execute(insert_rows)
    #conn.commit() # don't know when I would need to commit

conn.close()

# Attempt 3 # From a related Stack Exchange Post
 create the table but first drop if it already exists
command = """DROP TABLE IF EXISTS test_table
CREATE TABLE test_table # these columns are from my real dataset
"Serial Number" serial primary key,
"Dealer Code" text,
"FSHIP_DT" timestamp without time zone,
;"""
cursor.execute(command)
connection.commit()

# stream the data using 'to_csv' and StringIO(); then use sql's 'copy_from' function
output = io.StringIO()
# ignore the index
take_rates.to_csv(output, sep='~', header=False, index=False)
# jump to start of stream
output.seek(0)
contents = output.getvalue()
cur = connection.cursor()
# null values become ''
cur.copy_from(output, 'Config_Take_Rates_TEST', null="")
connection.commit()
cur.close()

在我看来，MS SQL Server 并不是一个很好的数据库......我想为粗略的格式表示歉意 - 我已经研究这个脚本几个星期了，但最终决定尝试为 StackOverflow 组织一些东西。非常感谢您提供的任何帮助!

Answer 1

如果您只需要替换现有表，请将其截断并使用 bcp 实用程序上传表。速度快得多。

from subprocess import call

command = "TRUNCATE TABLE test_table"
take_rates.to_csv('take_rates.csv', sep='\t', index=False)
call('bcp {t} in {f} -S {s} -U {u} -P {p} -d {db} -c -t "{sep}" -r "{nl}" -e {e}'.format(t='test_table', f='take_rates.csv', s=server, u=user, p=password, db=database, sep='\t', nl='\n')

您需要安装 bcp 实用程序(在 CentOS/RedHat 上安装 yum install mssql-tools)。