如何使用批量插入从Dataframe将数据插入到SQL Server表中？

Question

我是Python的新手，希望寻求帮助。我有一个在S3存储桶中的csv文件，我想使用Python pyodbc将这个csv文件导入到SQL Server的表中。这个文件大小为50 MB（400k条记录）。我的代码如下。如下所述，我的csv数据在一个dataframe中，如何使用Bulk insert将dataframe数据插入到sql server表中。如果我的方法不起作用，请给我提供另一种方法。

# Connection to S3
s3 = boto3.client(
    service_name = 's3',
    region_name = 'us-gov-west-1',
    aws_access_key_id = 'ZZZZZZZZZZZZZZZZZZ',
    aws_secret_access_key = 'AAAAAAAAAAAAAAAAA')

# Connection to SQL Server
server = 'myserver.amazonaws.com'
path = 'folder1/folder2/folder3/myCSVFile.csv'

cnxn = pyodbc.connect('DRIVER={SQL Server};SERVER='+server+';DATABASE=DB-staging;UID=User132;PWD=XXXXXX')
    cursor = cnxn.cursor()
    
obj_sum = s3.get_object(Bucket = 'my_bucket', Key = path)
csv_data = pd.read_csv(obj_sum['Body'])
df = pd.DataFrame(csv_data, columns = ['SYSTEM_NAME', 'BUCKET_NAME', 'LOCATION', 'FILE_NAME', 'LAST_MOD_DATE', 'FILE_SIZE'])
#print(df.head(n=15).to_string(index=False))

# Insert DataFrame to table
cursor.execute("""truncate table dbo.table1""")
cursor.execute("""BULK INSERT dbo.table1 FROM """ + .....# what do I put here since data is in dataframe??)

我尝试循环遍历数据框架，插入5k条记录花费了20分钟。以下是代码。遍历每条记录是一种选择，但不是好的选择。这就是为什么我如果可能的话会转向批量插入。

for i in df.itertuples(index = False):
    if i.FILE_SIZE != 0:
       cursor.execute("""insert into dbo.table1 (SYSTEM_NAME, BUCKET_NAME, X_LOCATION, FILE_NAME, LAST_MOD_DATE, FILE_SIZE) 
                  values (?,?,?,?,?,?)""", i.SYSTEM_NAME, i.BUCKET_NAME, i.LOCATION, i.FILE_NAME, i.LAST_MOD_DATE, i.FILE_SIZE)

最后，有一个额外的问题，我希望检查数据框中的“FILE_SIZE”列是否等于0，如果是则跳过该记录并继续处理下一条记录。

谢谢您的提前帮助。

- AJR

---

我需要使用turbodbc和pyarrow吗？创建一个Arrow表而不是DataFrame？ - AJR

为什么不使用to_sql呢？它使用多行插入，应该非常快。您是否进行了基准测试并发现速度太慢？ - Nick ODell

是的。我试过了，我的工作仍在运行，已经持续了9个小时，处理的文件大小为50MB。 - AJR

相关答案请点击此处。 - Gord Thompson

请参考此答案进行行过滤。 - Gord Thompson

Answer 1

感谢帮助。

使用 fast_executemany = True 对我很有帮助。

engine = sal.create_engine("mssql+pyodbc://username:password@"+server+":1433/db-name?driver=ODBC+Driver+17+for+SQL+Server?Trusted_Connection=yes",
                           fast_executemany = True)
conn = engine.connect()

我不得不修改我的代码来使用"sqlalchemy"，但现在它运行得非常好。

调用将数据上传到SQL Server的函数如下：

df.to_sql(str, con = engine, index = False, if_exists = 'replace')

- AJR

---

Sal 到底是指什么？ - Akshay Gupta

表格“表格名称”已经存在。 - vinsent paramanantham

@AkshayGupta 可能会使用 import sqlalchemy as sal。 - SamyIshak