mssql-python으로 데이터 로딩 및 이동 패턴을 선택하세요

이 드라이버는 mssql-python Microsoft SQL에 데이터를 쓰기 위한 여러 경로를 제공합니다. 각 경로는 서로 다른 업무량에 맞습니다. 이 가이드는 데이터 양, 소스 형식, 최신 의미론을 바탕으로 적합한 제품을 선택하는 데 도움을 줍니다.

업무량에 따라 결정하세요

업무량	권장 경로	이유
CSV 파일을 테이블에 로드하세요	CSV 데이터를 대량 복사본으로 로드하세요	bulkcopy() 생성기를 사용하면 어떤 크기의 파일이든 메모리에 로드하지 않고도 처리할 수 있습니다.
애플리케이션 코드에서 단일 행을 삽입합니다	단일 행 인서트	낮은 오버헤드와 직관적인 오류 처리, 생성된 키를 반환하는 데 OUTPUT과 함께 작동합니다.
애플리케이션 코드에서 소규모~중간 규모의 배치를 삽입하세요	일괄 삽입	단일 인서트에 비해 왕복 횟수가 줄어듭니다.
어떤 소스에서든 수백 행 이상 불러오세요	대량 복사	TDS 벌크 인서트는 대량 생산에 가장 효율적인 경로입니다.
키를 기반으로 행을 삽입하거나 업데이트합니다	MERGE으로 업서트	MERGE, INSERTUPDATE, 를 DELETE 하나의 명제로 처리합니다.
데이터프레임을 테이블에 로드합니다	데이터프레임 로드	bulkcopy_arrow()모든 값에 대해 Python 객체를 만들지 않고도 DataFrame의 Arrow 데이터를 읽습니다.
Apache Arrow 데이터를 테이블에 로드하기	애로우 데이터 로드	bulkcopy_arrow()Python 튜플을 만들지 않고 Arrow 메모리를 직접 읽습니다.
Parquet 파일로 데이터 준비	파르케트 스테이징	중간 파일 형식이 필요한 크로스 시스템 ETL에 유용합니다. Parquet은 이미 Arrow 데이터이기 때문에 행 변환 없이 로드됩니다.

대량 복사를 사용하여 CSV 데이터 로드

CSV 데이터를 로드하는 것이 Python 데이터베이스 작업에서 가장 흔한 인제스트 질문입니다. csv.reader에 전원을 공급하는 발전기와 함께 bulkcopy() 사용:

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()

대량 카피 성능 팁

• 대용량 데이터셋에는 생성기를 사용해 메모리 사용량을 일정하게 유지하세요.
• 소스가 DataFrame이나 Parquet 파일과 같은 열 기반 형식인 경우 bulkcopy_arrow()을 사용하세요. Python 행 튜플로의 변환을 건너뛸 수 있습니다.
• TDS 배치당 전송되는 행 수를 제어하려면 batch_size을 설정합니다. 5,000개로 시작해서 행 너비에 따라 조정하세요.
• 독점 부하에 테이블 락을 사용하세요: cursor.bulkcopy("dbo.ProductImport", rows, table_lock=True).
• 로드 전에 인덱스를 비활성화하고, 그 후에 다시 빌드하세요. 이 순서는 부하 시 인덱스 유지 오버헤드를 피할 수 있습니다.

열 매핑, 식별 열, NULL 처리, 병렬 로딩에 대해서는 벌크 복사 작업을 참조하세요.

업서트에 MERGE 사용

MERGE이는 Microsoft SQL의 조건부 INSERT, , UPDATEDELETE 에 대한 단일 연산에 대한 문장입니다. 이는 Python 개발자들에게 자주 필요한 "새로우면 삽입하고, 이미 존재하면 업데이트하는" 패턴을 처리합니다.

단일 행 업서트

단일 행에는 매개변수 별칭을 정의하는 MERGE 절과 함께 USING를 사용하세요:

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()

데이터프레임 로드

DataFrame은 열 기반이므로, bulkcopy_arrow()에 사용하려고 행 튜플로 평탄화하지 말고 bulkcopy()로 로드하세요.

로딩 전에 Arrow 타입을 목적지 열과 맞추세요. pyarrow운전자가 돈, 소수점, 숫자로 매핑할 수 없는 숫자 열을 추론합니다float64.

pandas

import pandas as pd
import pyarrow as pa

df = pd.read_csv("products.csv")

target = pa.schema([
    pa.field("Name", pa.string()),
    pa.field("ProductNumber", pa.string()),
    pa.field("ListPrice", pa.decimal128(19, 4)),   # MONEY
])

table = pa.Table.from_pandas(
    df[["Name", "ProductNumber", "ListPrice"]], preserve_index=False
).cast(target)

cursor.bulkcopy_arrow("dbo.ProductImport", table)
conn.commit()

스키마를 Table.from_pandas()에 전달하는 대신 Table.cast()를 사용하세요. Table.from_pandas()는 float 열을 decimal128로 직접 변환할 수 없습니다.

극지

Polars는 Arrow C 데이터 인터페이스를 구현하므로, DataFrame 자체를 전달할 수 있습니다. 같은 이유로 먼저 기둥을 주조하세요:

import polars as pl

df = pl.read_csv("products.csv")

cursor.bulkcopy_arrow(
    "dbo.ProductImport",
    df.select([
        "Name",
        "ProductNumber",
        pl.col("ListPrice").cast(pl.Decimal(19, 4)),   # MONEY
    ]),
)
conn.commit()

파일을 읽을 때 타입을 설정할 수도 있는데, pl.read_csv("products.csv", schema_overrides={"ListPrice": pl.Decimal(19, 4)}).

DataFrame을 전달하면 버퍼가 복제 없이 드라이버에 직접 전달됩니다. df.to_arrow() 이 방법도 작동하지만, Polars는 변환 과정에서 문자열 열을 다시 인코딩하여 모든 문자열 데이터를 복사합니다.

bulkcopy_arrow()는 RecordBatch, pyarrow.Table, __arrow_c_array__ 또는 __arrow_c_stream__나 RecordBatchReader를 통해 Arrow C 데이터 인터페이스를 구현하는 모든 객체를 받습니다. 이들 중 어느 것을 TypeError에 전달해도 bulkcopy()이 발생합니다.

완전한 데이터프레임 로딩 패턴은 판다스 적분 과 폴라스 적분을 참조하세요.

애로우 데이터 로드

소스가 이미 Apache Arrow 형식일 때는 cursor.bulkcopy_arrow() Python 튜플을 먼저 빌드하지 않고 불러와야 합니다.

from decimal import Decimal

import pyarrow as pa

# bulkcopy_arrow() opens its own connection, so commit the table creation first.
conn.autocommit = True
cursor = conn.cursor()

table = pa.table({
    "Name": pa.array(["Widget", "Gadget"], type=pa.string()),
    "ProductNumber": pa.array(["WI-1000", "GA-2000"], type=pa.string()),
    "ListPrice": pa.array([Decimal("29.99"), Decimal("49.99")], type=pa.decimal128(10, 2)),
})

result = cursor.bulkcopy_arrow("dbo.ProductImport", table, batch_size=5000)
print(f"Copied {result['rows_copied']} rows")

이 메서드는 a pyarrow.RecordBatch 또는 a pyarrow.RecordBatchReader도 받아들여서 결과 집합 cursor.arrow_reader() 을 다른 테이블로 바로 스트리밍할 수 있습니다.

각 Arrow 열 유형은 목적지 SQL 열 유형과 호환되어야 하며, 작성자는 유형 군 간 변환을 하지 않습니다. 자세한 내용은 Apache Arrow 통합을 참조하세요.

파르케트 스테이징

시스템 간 데이터 마이그레이션이나 ETL 파이프라인이 이미 Parquet 파일을 생성할 때 중간 형식으로 Parquet을 사용하세요. Parquet 파일을 읽으면 Arrow 데이터가 되므로 바로 bulkcopy_arrow()에 전달하세요:

import pyarrow.parquet as pq

cursor.bulkcopy_arrow("dbo.ProductImport", pq.read_table("products.parquet"))
conn.commit()

큰 Parquet 파일의 경우, 메모리 사용량을 일정하게 유지하기 위해 행 그룹을 반복합니다. 각 배치는 RecordBatch이며, bulkcopy_arrow()가 이를 직접 수용합니다:

import pyarrow.parquet as pq

parquet_file = pq.ParquetFile("products.parquet")

for batch in parquet_file.iter_batches(batch_size=10000):
    cursor.bulkcopy_arrow("dbo.ProductImport", batch)

conn.commit()

전체 파일을 한 번의 호출로 스트리밍하려면 배치를 다음과 같이 RecordBatchReader랩하세요:

import pyarrow as pa
import pyarrow.parquet as pq

parquet_file = pq.ParquetFile("products.parquet")
reader = pa.RecordBatchReader.from_batches(
    parquet_file.schema_arrow, parquet_file.iter_batches(batch_size=10000)
)

cursor.bulkcopy_arrow("dbo.ProductImport", reader)
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

관련 콘텐츠

• mssql-python과 함께 대량 복사를 사용하세요
• pssql-python과 pandas를 사용하세요
• Polars와 함께 mssql-python을 사용하세요
• mssql-python 애플리케이션을 위한 성능 튜닝
• mssql-python을 이용한 트랜잭션 관리
• mssql-python으로 데이터 접근 및 분석 패턴을 선택하세요