我使用 python 使用 Pandas 讀取 CSV，修復一些字段，然后將數據逐行寫入 SQL Server 中的表。服務器上禁用批量導入 - 另外，因為最終會有數十個這樣的文件，以自動下載和攝取文件。我可以看到這需要幾分鐘，但運行需要幾個小時。我知道如果啟用的話，我可以在幾秒鐘內批量上傳這些內容，但這可能是不可能的。問題是使用 python 每次運行可能需要 1 到 3 個小時。這是不可接受的。我想知道是否有更快的方法來完成此上傳。我可以對表格做些什么來加快導入速度，或者采用不同的編碼方式嗎？這是我正在使用的代碼類型的示例：def ingest_glief_reporting_exceptions_csv(): global conn global cursor filename = r"20200824-0800-gleif-goldencopy-repex-golden-copy.csv" # filename = r"repex_1K.csv" full_filename = os.path.join(raw_data_dir, filename) sql_str = "exec dbo.util_get_gleif_last_reporting_exception" cursor.execute(sql_str) last_lei = '' for result in cursor.fetchall(): last_lei = result[0] # "repex" is short for "reporting exceptions", shorten the headers repex_headers = [ 'LEI', 'ExceptionCategory', 'ExceptionReason1', 'ExceptionReason2', 'ExceptionReason3', 'ExceptionReason4', 'ExceptionReason5', 'ExceptionReference1', 'ExceptionReference2', 'ExceptionReference3', 'ExceptionReference4', 'ExceptionReference5' ] df = pd.read_csv(full_filename, header=0, quotechar='"') # Change to the column headers generated in VBA df.columns = repex_headers for colname in df.columns: df[colname] = df[colname].astype(str) df[colname] = df[colname].replace({'nan': ''}) place_holder = '?,?' for i in range(1, len(repex_headers)): place_holder += ',?' sql_str = "exec save_gleif_reporting_exception " + place_holder row_count = 0 row = dict() do_not_upload = True if last_lei == '': do_not_upload = False # There was no last uploaded record, so we can start now for index, row in df.iterrows(): row_count += 1 if do_not_upload: if row['LEI'] == last_lei: do_not_upload = False continue else: continue )
2 回答
慕絲7291255
TA貢獻1859條經驗 獲得超6個贊
由于您不需要存儲過程的返回值，因此您應該能夠使用 pandas 的to_sql方法將行直接插入表中。這段代碼...
from time import time
import pandas as pd
import sqlalchemy as sa
from_engine = sa.create_engine("mssql+pyodbc://@mssqlLocal64")
to_engine = sa.create_engine(
"mssql+pyodbc://sa:[email protected]/mydb"
"?driver=ODBC+Driver+17+for+SQL+Server",
fast_executemany=False,
)
# set up test
to_cnxn = to_engine.raw_connection()
to_cnxn.execute("TRUNCATE TABLE MillionRows")
to_cnxn.commit()
num_rows_to_upload = 10000
df = pd.read_sql_query(
f"SELECT TOP {num_rows_to_upload} "
"[TextField], [LongIntegerField], [DoubleField], [varchar_column] "
"FROM MillionRows ORDER BY ID",
from_engine,
)
# run test
t0 = time()
df.to_sql("MillionRows", to_engine, index=False, if_exists="append")
s = f"{(time() - t0):0.1f} seconds"
print(f"uploading {num_rows_to_upload:,d} rows took {s}")
… 表示與您現在所做的工作大致相同的內部工作水平，即，將每個單獨的行作為單獨的調用上傳.execute。結果是
uploading 10,000 rows took 60.2 seconds
但是，簡單地更改to_engine為使用fast_executemany=True結果
uploading 10,000 rows took 1.4 seconds
侃侃爾雅
TA貢獻1801條經驗 獲得超16個贊
關閉自動提交
?conn = pyodbc.connect(
? ? ? ? r'DRIVER={ODBC Driver 17 for SQL Server};SERVER=' + server_name + '; \
? ? ? ? Database=' + db_name + ';Trusted_Connection=yes;', timeout=5, autocommit=False)
?
并在此處和循環結束時提交。
? ? if index % 1000 == 0:
? ? ? ? ? ? print("Imported %s rows" % (index))
使用自動提交，您必須等待日志文件在每一行之后保存到磁盤。
為了進一步優化，如果您使用 SQL 2016+，請使用 JSON 將批量行發送到 SQL Server，并使用OPENJSON在服務器端進行解析。
添加回答
舉報
0/150
提交
取消