이 드라이버는 mssql-python Microsoft SQL에 데이터를 쓰기 위한 여러 경로를 제공합니다. 각 경로는 서로 다른 업무량에 맞습니다. 이 가이드는 데이터 양, 소스 형식, 최신 의미론을 바탕으로 적합한 제품을 선택하는 데 도움을 줍니다.
업무량에 따라 결정하세요
| 업무량 | 권장 경로 | 이유 |
|---|---|---|
| CSV 파일을 테이블에 로드하세요 | CSV 데이터를 대량 복사본으로 로드하세요 |
bulkcopy() 생성기를 사용하면 어떤 크기의 파일이든 메모리에 로드하지 않고도 처리할 수 있습니다. |
| 애플리케이션 코드에서 단일 행을 삽입합니다 | 단일 행 인서트 | 낮은 오버헤드와 직관적인 오류 처리, 생성된 키를 반환하는 데 OUTPUT과 함께 작동합니다. |
| 애플리케이션 코드에서 소규모~중간 규모의 배치를 삽입하세요 | 일괄 삽입 | 단일 인서트에 비해 왕복 횟수가 줄어듭니다. |
| 어떤 소스에서든 수백 행 이상 불러오세요 | 대량 복사 | TDS 벌크 인서트는 대량 생산에 가장 효율적인 경로입니다. |
| 키를 기반으로 행을 삽입하거나 업데이트합니다 | MERGE으로 업서트 |
MERGE, INSERTUPDATE, 를 DELETE 하나의 명제로 처리합니다. |
| 데이터프레임을 테이블에 로드합니다 | 데이터프레임 로드 | pandas 또는 Polars에서 행을 추출하여 bulkcopy()에 전달합니다. |
| Parquet 파일로 데이터 준비 | 파르케트 스테이징 | 중간 파일 형식이 필요한 크로스 시스템 ETL에 유용합니다. |
대량 복사를 사용하여 CSV 데이터 로드
CSV 데이터를 로드하는 것이 Python 데이터베이스 작업에서 가장 흔한 인제스트 질문입니다.
bulkcopy()에 전원을 공급하는 발전기와 함께 csv.reader 사용:
import csv
import mssql_python
conn = mssql_python.connect(connection_string)
cursor = conn.cursor()
# Create a target table
cursor.execute("""
IF NOT EXISTS (SELECT * FROM sys.tables WHERE name = 'ProductImport')
CREATE TABLE dbo.ProductImport (
Name nvarchar(100),
ProductNumber nvarchar(25),
ListPrice decimal(10,2)
)
""")
conn.commit()
def csv_rows(path):
with open(path, newline="", encoding="utf-8") as f:
reader = csv.reader(f)
next(reader) # Skip header
for row in reader:
yield (row[0], row[1], float(row[2]))
result = cursor.bulkcopy(
"dbo.ProductImport",
csv_rows("products.csv"),
batch_size=5000
)
print(f"Loaded {result['rows_copied']} rows")
conn.commit()
생성기 패턴은 파일 크기와 상관없이 메모리 사용량을 일정하게 유지합니다. 컬럼 매핑과 정체성 처리에 대해서는 벌크 복사 작업을 참조하세요.
단일 행 인서트
애플리케이션 레벨 쓰기에는 단일 인서트를 사용하여 한 번에 한 레코드씩 처리하세요. 생성된 키를 가져오는 방법 OUTPUT INSERTED :
cursor.execute("""
INSERT INTO dbo.ProductImport (Name, ProductNumber, ListPrice)
OUTPUT INSERTED.Name
VALUES (%(name)s, %(product_number)s, %(list_price)s)
""", {"name": "Widget", "product_number": "WG-1000", "list_price": 19.99})
inserted_name = cursor.fetchval()
conn.commit()
단일 인서트는 다음과 같은 상황에서 적합한 선택입니다:
- 사용자 동작(폼 제출, API 호출)마다 한 행씩 삽입합니다.
- 삽입 전에 각 행을 개별적으로 검증하거나 변환해야 합니다.
- 삽입된 ID나 다른 생성된 값이 즉시 필요합니다.
일괄 삽입
행 수가 적당하고 대량 복사본의 처리량이 필요 없을 때 사용 executemany() 하세요:
rows = [
{"name": "Widget A", "product_number": "WG-1001", "list_price": 19.99},
{"name": "Widget B", "product_number": "WG-1002", "list_price": 24.99},
{"name": "Widget C", "product_number": "WG-1003", "list_price": 29.99},
]
cursor.executemany(
"INSERT INTO dbo.ProductImport (Name, ProductNumber, ListPrice) VALUES (%(name)s, %(product_number)s, %(list_price)s)",
rows
)
conn.commit()
executemany() 각 행을 별도의 매개변수화된 문으로 보냅니다. 처리량이 행당 제어보다 더 중요할 때는 bulkcopy() TDS 벌크 인서트 프로토콜을 사용하기 때문에 더 효율적입니다. 크로스오버는 행 너비와 네트워크 지연 시간에 따라 다르지만, 보통 수백 행 초반 정도입니다.
대량 복사
처리량이 행별 제어보다 더 중요하다면 bulkcopy()를 사용하세요. 이 프로토콜은 TDS 벌크 인서트 프로토콜을 사용하며, 행별 인서트보다 훨씬 효율적입니다:
rows = [
("Widget A", "WG-1001", 19.99),
("Widget B", "WG-1002", 24.99),
("Widget C", "WG-1003", 29.99),
]
result = cursor.bulkcopy("dbo.ProductImport", rows, batch_size=5000)
print(f"Loaded {result['rows_copied']} rows")
conn.commit()
대량 카피 성능 팁
- 대용량 데이터셋에는 생성기를 사용해 메모리 사용량을 일정하게 유지하세요.
- TDS 배치당 전송되는 행 수를 제어하려면
batch_size을 설정합니다. 5,000개로 시작해서 행 너비에 따라 조정하세요. - 독점 부하에 테이블 락을 사용하세요:
cursor.bulkcopy("dbo.ProductImport", rows, table_lock=True). - 로드 전에 인덱스를 비활성화하고, 그 후에 다시 빌드하세요. 이 순서는 부하 시 인덱스 유지 오버헤드를 피할 수 있습니다.
열 매핑, 식별 열, NULL 처리, 병렬 로딩에 대해서는 벌크 복사 작업을 참조하세요.
업서트에 MERGE 사용
MERGE이는 Microsoft SQL의 조건부 INSERT, , UPDATEDELETE 에 대한 단일 연산에 대한 문장입니다. 이는 Python 개발자들에게 자주 필요한 "새로우면 삽입하고, 이미 존재하면 업데이트하는" 패턴을 처리합니다.
단일 행 업서트
단일 행에는 매개변수 별칭을 정의하는 USING 절과 함께 MERGE를 사용하세요:
cursor.execute("""
MERGE dbo.ProductImport AS target
USING (SELECT %(name)s AS Name, %(product_number)s AS ProductNumber, %(list_price)s AS ListPrice) AS source
ON target.ProductNumber = source.ProductNumber
WHEN MATCHED THEN
UPDATE SET
Name = source.Name,
ListPrice = source.ListPrice
WHEN NOT MATCHED THEN
INSERT (Name, ProductNumber, ListPrice)
VALUES (source.Name, source.ProductNumber, source.ListPrice);
""", {"name": "Widget A", "product_number": "WG-1001", "list_price": 24.99})
conn.commit()
스테이징 테이블이 있는 벌크 업서트
대량 업서트의 경우, 데이터를 먼저 임시 테이블에 스테이징한 뒤 MERGE 그걸로 업데이트하세요. DataFrame 업서트 및 배치 업데이트의 기본 패턴으로 insert-or-update를 사용하세요:
import csv
import mssql_python
conn = mssql_python.connect(connection_string)
cursor = conn.cursor()
# Step 1: Create a global temp table for staging
# Note: bulkcopy() requires global temp tables (##), not session temp tables (#)
cursor.execute("""
IF OBJECT_ID('tempdb..##ProductImportStage') IS NOT NULL
DROP TABLE ##ProductImportStage;
CREATE TABLE ##ProductImportStage (
Name nvarchar(100),
ProductNumber nvarchar(25),
ListPrice decimal(10,2)
)
""")
cursor.commit()
# Step 2: Bulk load into the staging table
def csv_rows(path):
with open(path, newline="", encoding="utf-8") as f:
reader = csv.reader(f)
next(reader)
for row in reader:
yield (row[0], row[1], float(row[2]))
cursor.bulkcopy("##ProductImportStage", csv_rows("products_update.csv"), batch_size=5000)
# Step 3: MERGE from staging into the target table
cursor.execute("""
MERGE dbo.ProductImport AS target
USING ##ProductImportStage AS source
ON target.ProductNumber = source.ProductNumber
WHEN MATCHED THEN
UPDATE SET
Name = source.Name,
ListPrice = source.ListPrice
WHEN NOT MATCHED BY TARGET THEN
INSERT (Name, ProductNumber, ListPrice)
VALUES (source.Name, source.ProductNumber, source.ListPrice)
OUTPUT $action, INSERTED.ProductNumber, DELETED.ProductNumber;
""")
# Step 4: Read the OUTPUT to see what changed
for row in cursor.fetchall():
print(f"{row[0]}: inserted={row[1]}, deleted={row[2]}")
conn.commit()
이 예시는 기본 삽입 또는 업데이트 패턴을 보여줍니다:
-
INSERT 소스에서 타겟
WHEN NOT MATCHED BY TARGET에 존재하지 않는 행들(). -
UPDATE 두 행 모두에 존재하는 행들(
WHEN MATCHED). - OUTPUT 절은 각 행에서 어떤 조치가 취해졌는지 보고하며, 이는 감사 추적에 유용합니다.
주의
스테이징 데이터가 타겟의 권위 있는 전체 스냅샷일 때만 추가 WHEN NOT MATCHED BY SOURCE THEN DELETE 하세요. 배치에 변경된 행만 포함되어 있다면, 해당 절은 소스 피드에서 의도적으로 생략된 행을 삭제합니다.
전체 조정이 필요한 경우, 원본이 대상 테이블의 신뢰할 수 있는 기준임을 확인한 후에만 MERGE을 확장하세요:
WHEN NOT MATCHED BY SOURCE THEN
DELETE
공유 환경에서는 각 실행마다 고유한 글로벌 임시 테이블 이름이나 영구 스테이징 테이블을 사용하여 동시 작업 간 충돌을 방지합니다.
언제 separate UPDATE and INSERT 문장을 대신 사용해야 할까요
MERGE 강력하지만 예외적인 경우도 있습니다. 다음과 같은 경우에는 별도의 문장을 사용하는 것을 고려해 보세요:
-
DELETE 논리는 필요하지 않아요. 별도의
UPDATE뒤에 이어지는INSERT WHERE NOT EXISTS것이 더 읽기 쉽고 디버깅도 쉽습니다. - 이 명제는
MERGE너무 복잡해서 잠금 동작을 예측하기 어렵습니다. 별도의 문장은 잠금 세분화에 대한 명확한 제어를 제공합니다. - 고동시성 테이블을 업데이트하는 중인데,
MERGE락 에스컬레이션이 블로킹을 일으킬 수 있습니다.
# Simpler alternative: UPDATE then INSERT
cursor.execute("""
UPDATE dbo.ProductImport
SET Name = %(name)s, ListPrice = %(list_price)s
WHERE ProductNumber = %(product_number)s
""", {"name": "Widget A", "list_price": 24.99, "product_number": "WG-1001"})
if cursor.rowcount == 0:
cursor.execute("""
INSERT INTO dbo.ProductImport (Name, ProductNumber, ListPrice)
VALUES (%(name)s, %(product_number)s, %(list_price)s)
""", {"name": "Widget A", "product_number": "WG-1001", "list_price": 24.99})
conn.commit()
데이터프레임 로드
판다스나 폴라스 데이터프레임에서 행을 추출하여 다음을 사용 bulkcopy()해 불러오세요:
pandas
pandas DataFrame을 튜플로 변환하여 bulkcopy()에 전달:
import pandas as pd
df = pd.read_csv("products.csv")
# Convert DataFrame rows to tuples
rows = list(df[["Name", "ProductNumber", "ListPrice"]].itertuples(index=False, name=None))
cursor.bulkcopy("dbo.ProductImport", rows, batch_size=5000)
conn.commit()
극지
Polars 데이터프레임을 튜플로 변환하는 방법은 다음과 같습니다 .rows() :
import polars as pl
df = pl.read_csv("products.csv")
# Convert Polars DataFrame to list of tuples
rows = df.select(["Name", "ProductNumber", "ListPrice"]).rows()
cursor.bulkcopy("dbo.ProductImport", rows, batch_size=5000)
conn.commit()
완전한 데이터프레임 로딩 패턴은 판다스 적분 과 폴라스 적분을 참조하세요.
파르케트 스테이징
시스템 간 데이터 마이그레이션이나 ETL 파이프라인이 이미 Parquet 파일을 생성할 때 중간 형식으로 Parquet을 사용하세요:
import pyarrow.parquet as pq
# Read Parquet file
table = pq.read_table("products.parquet")
# Convert to rows for bulkcopy
rows = [tuple(row) for row in zip(*[col.to_pylist() for col in table.columns])]
cursor.bulkcopy("dbo.ProductImport", rows, batch_size=5000)
conn.commit()
큰 Parquet 파일의 경우, 메모리 사용량을 일정하게 유지하기 위해 행 그룹을 읽으세요:
import pyarrow.parquet as pq
parquet_file = pq.ParquetFile("products.parquet")
for batch in parquet_file.iter_batches(batch_size=10000):
rows = [tuple(row) for row in zip(*[col.to_pylist() for col in batch.columns])]
cursor.bulkcopy("dbo.ProductImport", rows, batch_size=10000)
conn.commit()
로드된 데이터 검증
로딩 후에는 행 수와 스팟 체크 데이터를 확인하세요:
cursor.execute("SELECT COUNT(*) FROM dbo.ProductImport")
count = cursor.fetchval()
print(f"Total rows: {count}")
cursor.execute("""
SELECT TOP 5 Name, ProductNumber, ListPrice
FROM dbo.ProductImport
ORDER BY Name
""")
for row in cursor:
print(f" {row.Name} ({row.ProductNumber}): ${row.ListPrice:.2f}")
프로덕션 환경의 부하에서는 bulkcopy() 호출을 보호하기 위해 호출하는 연결의 트랜잭션에 의존하지 마세요.
bulkcopy() 내부 연결을 열고 복사한 행을 독립적으로 커밋하기 때문에, 메인 연결의 A conn.rollback() 는 이를 되돌릴 수 없습니다. 원자성을 부여하는 두 가지 접근법이 있습니다:
-
use_internal_transaction=True를 각 배치가 자체 트랜잭션으로 처리되도록 설정하세요. 중간에 실패한 배치는 반쯤 부하된 상태로 두지 않고 그 배치를 롤백합니다. - 데이터를 반영하기 전에 유효성을 검사하려면 데이터를 스테이징 테이블에 대량 복사한 다음 유효성을 검사한 후, 주 연결에서 트랜잭션 내
INSERT ... SELECT을 사용하여 행을 대상 테이블로 이동하세요. 그것은INSERT사용자의 연결에서 실행되므로conn.rollback()검증에 실패하면 이를 취소합니다.
# Stage the data. bulkcopy() runs on its own connection, so these rows
# persist regardless of the transaction below.
cursor.bulkcopy("dbo.ProductImport_Stage", rows, batch_size=5000)
try:
cursor.execute("SELECT COUNT(*) FROM dbo.ProductImport_Stage")
count = cursor.fetchval()
if count < expected_count:
raise ValueError(f"Expected {expected_count} rows, got {count}")
# This INSERT runs on your connection, so it's covered by the transaction.
cursor.execute("""
INSERT INTO dbo.ProductImport (Name, ProductNumber, ListPrice)
SELECT Name, ProductNumber, ListPrice FROM dbo.ProductImport_Stage
""")
conn.commit()
except Exception:
conn.rollback()
raise