使用 python 将 csv 文件写入 SQL Server 数据库

Question

我正在尝试使用 python 将 csv 文件写入 SQL Server 数据库中的表中。传递参数时我遇到错误，但是手动执行时我没有遇到任何错误。这是我正在执行的代码。

cur=cnxn.cursor() # Get the cursor
csv_data = csv.reader(file(Samplefile.csv')) # Read the csv
for rows in csv_data: # Iterate through csv
    cur.execute("INSERT INTO MyTable(Col1,Col2,Col3,Col4) VALUES (?,?,?,?)",rows)
cnxn.commit()

错误：

pyodbc.DataError: ('22001', '[22001] [Microsoft][ODBC SQL Server Driver][SQL Server]字符串或二进制数据将被截断。(8152) (SQLExecDirectW); [01000] [Microsoft] [ODBC SQL Server Driver][SQL Server]语句已终止。(3621)')

但是，当我手动插入值时。效果很好

cur.execute("INSERT INTO MyTable(Col1,Col2,Col3,Col4) VALUES (?,?,?,?)",'A','B','C','D')

我已确保 TABLE 在数据库中，数据类型与我传递的数据一致。连接和光标也正确。行的数据类型是“列表”

Answer 1

考虑动态构建查询以确保占位符的数量与您的表格和 CSV 文件格式相匹配。然后，只需确保您的表格和 CSV 文件正确无误，而不是检查您是否在代码中输入了足够多的 ? 占位符。

以下示例假设

1. CSV 文件在第一行包含列名
2. 已建立连接
3. 文件名为test.csv
4. 表名是MyTable
5. Python 3

...
with open ('test.csv', 'r') as f:
    reader = csv.reader(f)
    columns = next(reader) 
    query = 'insert into MyTable({0}) values ({1})'
    query = query.format(','.join(columns), ','.join('?' * len(columns)))
    cursor = connection.cursor()
    for data in reader:
        cursor.execute(query, data)
    cursor.commit()

如果文件中不包含列名：

...
with open ('test.csv', 'r') as f:
    reader = csv.reader(f)
    data = next(reader) 
    query = 'insert into MyTable values ({0})'
    query = query.format(','.join('?' * len(data)))
    cursor = connection.cursor()
    cursor.execute(query, data)
    for data in reader:
        cursor.execute(query, data)
    cursor.commit()

Answer 2

我修改了上面由 Brian 编写的代码，因为上面发布的代码不适用于我尝试上传的分隔文件。 row.pop() 行也可以忽略，因为它仅适用于我尝试上传的文件集。

import csv

def upload_table(path, filename, delim, cursor):
    """
    Function to upload flat file to sqlserver
    """
    tbl = filename.split('.')[0]
    cnt = 0
    with open (path + filename, 'r') as f:
        reader = csv.reader(f, delimiter=delim)
        for row in reader:
            row.pop() # can be commented out
            row = ['NULL' if val == '' else val for val in row]
            row = [x.replace("'", "''") for x in row]
            out = "'" + "', '".join(str(item) for item in row) + "'"
            out = out.replace("'NULL'", 'NULL')
            query = "INSERT INTO " + tbl + " VALUES (" + out + ")"
            cursor.execute(query)
            cnt = cnt + 1
            if cnt % 10000 == 0:
                cursor.commit()
        cursor.commit()
    print("Uploaded " + str(cnt) + " rows into table " + tbl + ".")

Answer 3

如果你在气流中使用 MySqlHook，如果 cursor.execute() 带参数抛出 san 错误

TypeError：字符串格式化期间并非所有参数都转换

使用%s 代替?

with open('/usr/local/airflow/files/ifsc_details.csv','r') as csv_file:
    csv_reader = csv.reader(csv_file)
    columns = next(csv_reader)
    query = '''insert into ifsc_details({0}) values({1});'''
    query = query.format(','.join(columns), ','.join(['%s'] * len(columns)))
    mysql = MySqlHook(mysql_conn_id='local_mysql')
    conn = mysql.get_conn()
    cursor = conn.cursor()
    for data in csv_reader:
        cursor.execute(query, data)
    cursor.commit()

Answer 4

您可以将列作为参数传递。例如：

for rows in csv_data: # Iterate through csv
    cur.execute("INSERT INTO MyTable(Col1,Col2,Col3,Col4) VALUES (?,?,?,?)", *rows)

Answer 5

我把它整理好了。该错误是由于表的大小限制限制。它将列容量从 col1 varchar(10) 更改为 col1 varchar(35) 等。现在它工作正常。

Answer 6

您还可以使用以下任一方法将数据导入 SQL：

• SQL Server 导入和导出向导
• SQL Server 集成服务 (SSIS)
• OPENROWSET 函数

更多细节可以在这个网页上找到： https://docs.microsoft.com/en-us/sql/relational-databases/import-export/import-data-from-excel-to-sql?view=sql-server-2017

Answer 7

这是脚本，希望对你有用：

import pandas as pd    
import pyodbc as pc    
connection_string = "Driver=SQL Server;Server=localhost;Database={0};Trusted_Connection=Yes;"                           
cnxn = pc.connect(connection_string.format("DataBaseNameHere"), autocommit=True)    
cur=cnxn.cursor()
df= pd.read_csv("your_filepath_and_filename_here.csv").fillna('')    
query = 'insert into TableName({0}) values ({1})'    
query = query.format(','.join(df.columns), ','.join('?' * len(df1.columns)))        
cur.fast_executemany = True    
cur.executemany(query, df.values.tolist())
cnxn.close()