mssql-python과 함께 대량 복사를 사용하세요

mssql-python 드라이버에는 SQL Server, Azure SQL Database, Azure SQL Managed Instance, Microsoft Fabric의 SQL 데이터베이스에 대량의 데이터를 효율적으로 삽입하는 대량 복사 기능이 포함되어 있습니다.

이 cursor.bulkcopy() 방법은 대규모 데이터셋을 로드하는 데 있어 고성능 경로를 제공합니다:

네트워크 왕복 횟수를 최소화합니다.
선택적으로 로드 시 제약 검사를 우회할 수 있습니다.
최적화된 TDS 벌크 인서트 프로토콜을 사용합니다.
bcp.exe 및 SqlBulkCopy에 필적하는 처리량을 달성합니다.

Rust 기반 mssql_py_core 의 네이티브 확장 기능이 대량 복사 기능을 구동합니다. 일반 커서 execute() 파이프라인 밖에서 실행됩니다.

기본 사용법

커서에서 bulkcopy()를 호출하여 대상 테이블 이름과 행 튜플 또는 Row 객체로 이루어진 이터러블을 전달합니다:

중요합니다

같은 세션에서 대상 테이블을 생성하거나 변경하는 경우, bulkcopy() 전에 conn.commit()를 호출하세요. 벌크 복사 프로토콜은 별도의 내부 채널을 사용해 테이블 메타데이터를 읽기 때문에, 커밋되지 않은 DDL 변경은 교착 상태나 타임아웃을 일으킬 수 있습니다.

import mssql_python

conn = mssql_python.connect(connection_string)
cursor = conn.cursor()

# Create a temp table for the demo
cursor.execute("""
    CREATE TABLE ##BulkDemo (
        ID INT,
        Name NVARCHAR(50),
        Amount MONEY
    )
""")
conn.commit()

data = [
    (1, "Alice", 50000.00),
    (2, "Bob", 60000.00),
    (3, "Carol", 55000.00),
]

result = cursor.bulkcopy("##BulkDemo", data)
print(f"Copied {result['rows_copied']} rows in {result['batch_count']} batch(es)")
print(f"Elapsed: {result['elapsed_time']}")

반환 값

bulkcopy() 사전을 반환합니다:

암호키	Type	설명
`rows_copied`	int	복사 성공한 행 수.
`batch_count`	int	처리된 배치 수.
`elapsed_time`	float	작업에 걸린 시간(초)입니다.

메서드 서명

cursor.bulkcopy(
    table_name,                    # str – target table (can include schema, e.g. "dbo.MyTable")
    data,                          # Iterable[Tuple | Row] – rows to insert
    batch_size=0,                  # int – rows per batch; 0 = server optimal
    timeout=30,                    # int – operation timeout in seconds
    column_mappings=None,          # List[str] | List[Tuple[int,str]] | None
    keep_identity=False,           # bool – preserve identity values from source
    check_constraints=False,       # bool – check constraints during load
    table_lock=False,              # bool – use table-level lock
    keep_nulls=False,              # bool – preserve NULLs instead of defaults
    fire_triggers=False,           # bool – fire INSERT triggers on target
    use_internal_transaction=False, # bool – use internal transaction per batch
)

컬럼 매핑

기본적으로 bulkcopy() 열은 순서 위치별로 매핑됩니다. 각 데이터 열은 동일한 인덱스에 있는 테이블 열과 매핑됩니다. 매개변수를 column_mappings 사용해 이 동작을 무시하세요.

열 이름 목록

목록 내 각 위치는 출처 데이터 인덱스에 대응합니다:

result = cursor.bulkcopy(
    "##BulkDemo",
    data,
    column_mappings=["ID", "Name", "Amount"],
)

고급 형식: 명시적 인덱스 매핑

각 튜플은 의 형태 (source_index, target_column_name)를 취합니다. 이 형식을 사용하여 열을 건너뛰거나 재정렬하세요:

result = cursor.bulkcopy(
    "##BulkDemo",
    data,
    column_mappings=[(0, "ID"), (1, "Name"), (2, "Amount")],
)

파일에서 불러오기

생성기를 bulkcopy()전달하여 CSV 파일 및 기타 파일 형식에서 데이터를 불러올 수 있습니다.

CSV 파일

import csv
import io
import mssql_python

# In production, replace io.StringIO with open("data.csv", "r", ...)
csv_data = """ID,Name,Value
1,Widget,9.99
2,Gadget,24.50
3,Gizmo,4.75
"""

def csv_row_generator(file_obj):
    """Generator that yields tuples from a CSV file object."""
    reader = csv.reader(file_obj)
    next(reader)  # Skip header
    for row in reader:
        if row:  # skip blank lines
            yield (
                int(row[0]),      # ID
                row[1],           # Name
                float(row[2]),    # Value
            )

conn = mssql_python.connect(connection_string)
cursor = conn.cursor()

cursor.execute("""
    CREATE TABLE ##CSVImport (ID INT, Name NVARCHAR(100), Value FLOAT)
""")
conn.commit()
result = cursor.bulkcopy("##CSVImport", csv_row_generator(io.StringIO(csv_data)))
print(f"Imported {result['rows_copied']} rows from CSV")

배치가 있는 대형 파일

매개변수를 batch_size 배치당 드라이버가 보내는 행 수를 제어하도록 설정하세요. 이 방법은 큰 파일에 적합합니다:

import csv
import io
import mssql_python

# In production, replace io.StringIO with open("large_file.csv", "r", ...)
csv_data = "\n".join(
    ["ID,Name,Value"] + [f"{i},Item {i},{i * 1.5}" for i in range(1, 201)]
)

def csv_rows(file_obj):
    reader = csv.reader(file_obj)
    next(reader)  # Skip header
    for row in reader:
        if row:
            yield (int(row[0]), row[1], float(row[2]))

conn = mssql_python.connect(connection_string)
cursor = conn.cursor()

cursor.execute("""
    CREATE TABLE ##LargeCSV (ID INT, Name NVARCHAR(100), Value FLOAT)
""")
conn.commit()
result = cursor.bulkcopy(
    "##LargeCSV",
    csv_rows(io.StringIO(csv_data)),
    batch_size=50,
)
print(f"Imported {result['rows_copied']} rows in {result['batch_count']} batches")

pandas 데이터프레임 불러오기

bulkcopy()에 전달하기 전에 pandas DataFrame을 튜플 목록으로 변환합니다:

import pandas as pd
import mssql_python

df = pd.DataFrame({
    'ID': [1, 2, 3],
    'Name': ['Alice', 'Bob', 'Carol'],
    'Amount': [50000.0, 60000.0, 55000.0],
})

conn = mssql_python.connect(connection_string)
cursor = conn.cursor()

cursor.execute("""
    CREATE TABLE ##PandasDemo (ID INT, Name NVARCHAR(50), Amount MONEY)
""")
conn.commit()

data = [tuple(row) for row in df.itertuples(index=False, name=None)]
result = cursor.bulkcopy("##PandasDemo", data)

NULL 값 처리

어느 열 위치에든 None를 전달하여 SQL NULL 값을 삽입하세요:

cursor.execute("""
    CREATE TABLE ##NullDemo (ID INT, Name NVARCHAR(50), Amount MONEY)
""")
conn.commit()

data = [
    (1, "Alice", 50000.00),
    (2, "Bob", None),       # NULL Amount
    (3, None, 55000.00),    # NULL Name
]

cursor.bulkcopy("##NullDemo", data)

식별 열

명시적 ID 값을 삽입하려면 keep_identity=True:을(를) 설정하십시오.

cursor.execute("""
    CREATE TABLE ##IdentDemo (ID INT, Name NVARCHAR(50), Amount MONEY)
""")
conn.commit()

data = [
    (100, "Alice", 50000.00),
    (200, "Bob", 60000.00),
]

cursor.bulkcopy("##IdentDemo", data, keep_identity=True)

기본값인 keep_identity=False인 경우 데이터에서 identity 열을 생략하고 column_mappings을 사용하여 nonidentity 열을 대상으로 지정하세요.

대량 복사 옵션

매개 변수	Default	설명
`batch_size`	`0`	배치당 행 수. `0` 서버가 최적의 크기를 선택하게 하세요.
`timeout`	`30`	몇 초 후 작업 타임아웃.
`keep_identity`	`False`	원본 데이터의 신원 값을 보존하세요.
`check_constraints`	`False`	로드 중에 테이블 제약 조건을 확인하세요.
`table_lock`	`False`	행 레벨 잠금 대신 테이블 레벨 잠금장치를 획득하세요.
`keep_nulls`	`False`	열 기본값을 삽입하는 대신 NULL 값을 보존하세요.
`fire_triggers`	`False`	대상 테이블에서 INSERT 트리거를 실행합니다.
`use_internal_transaction`	`False`	각 배치를 내부 트랜잭션으로 묶으세요.

오류를 처리하십시오.

bulkcopy() 부하가 실패하면 예외를 발생시키므로, 호출을 블록 안에 try/except 감아 오류를 포착합니다. 참고로 bulkcopy()는 자체 내부 연결에서 실행되고 복사된 행을 독립적으로 커밋하므로, 메인 연결에서의 conn.rollback()로는 이를 되돌릴 수 없습니다. 배치를 원자적으로 처리하려면 use_internal_transaction=True를 설정하세요. 그러면 각 배치가 자체 트랜잭션으로 래핑되어, 배치가 실패할 경우 자동으로 롤백됩니다:

import mssql_python

conn = mssql_python.connect(connection_string)
cursor = conn.cursor()

cursor.execute("""
    CREATE TABLE ##ImportDemo (ID INT, Name NVARCHAR(50), Value FLOAT)
""")
conn.commit()

data = [
    (1, "Alice", 50000.00),
    (2, "Bob", 60000.00),
    (3, "Carol", 55000.00),
]

try:
    result = cursor.bulkcopy("##ImportDemo", data, use_internal_transaction=True)
    print(f"Successfully copied {result['rows_copied']} rows")
except (mssql_python.DatabaseError, ValueError) as e:
    # bulkcopy() commits on its own connection, so there's nothing to roll back
    # here. With use_internal_transaction=True, a failed batch is already rolled
    # back on the bulk copy connection.
    print(f"Bulk copy failed: {e}")

데이터 적재를 자체 검증 로직으로 제어하려면 먼저 스테이징 테이블에 대량 복사한 다음, 주 연결의 트랜잭션 내에서 INSERT ... SELECT을 사용해 해당 행을 대상 테이블로 옮기세요. 해당 작업은 사용자의 연결을 통해 INSERT 실행되므로, 유효성 검사에 실패하면 conn.rollback() 이를 실행 취소합니다.

Authentication

벌크 카피는 별도의 내부 채널을 사용하며, 이 채널은 별도의 토큰을 요구합니다. 드라이버는 지원되는 인증 방법에 대해 자동으로 토큰 획득을 처리합니다.

관리 신원 (ActiveDirectoryMSI)

시스템 할당 또는 사용자 할당 관리 ID에는 Authentication=ActiveDirectoryMSI를 사용합니다. 이 인증 방법은 Azure VMS, App Service, Functions, AKS와 같은 Azure 호스팅 서비스에 권장됩니다.

import mssql_python

# System-assigned managed identity
conn = mssql_python.connect(
    "Server=<server>.database.windows.net;"
    "Database=<database>;"
    "Authentication=ActiveDirectoryMSI;"
    "Encrypt=yes"
)
cursor = conn.cursor()

cursor.execute("CREATE TABLE ##MsiDemo (ID INT, Name NVARCHAR(50))")
conn.commit()

result = cursor.bulkcopy("##MsiDemo", [(1, "Alice"), (2, "Bob")])
print(f"Copied {result['rows_copied']} rows")

사용자 지정 관리 신원의 경우, 클라이언트 ID를 연결 문자열에 전달합니다:

conn = mssql_python.connect(
    "Server=<server>.database.windows.net;"
    "Database=<database>;"
    "Authentication=ActiveDirectoryMSI;"
    "UID=<client-id>;"
    "Encrypt=yes"
)

서비스 프린시펄 (ActiveDirectoryServicePrincipal)

서비스 주체(클라이언트 자격 증명) 인증에 사용됩니다 Authentication=ActiveDirectoryServicePrincipal .

conn = mssql_python.connect(
    "Server=<server>.database.windows.net;"
    "Database=<database>;"
    "Authentication=ActiveDirectoryServicePrincipal;"
    "UID=<application-client-id>;"
    "PWD=<client-secret>;"
    "Encrypt=yes"
)
cursor = conn.cursor()

cursor.execute("CREATE TABLE ##SpDemo (ID INT, Value FLOAT)")
conn.commit()

result = cursor.bulkcopy("##SpDemo", [(1, 1.5), (2, 2.5)])
print(f"Copied {result['rows_copied']} rows")

기본 자격 증명 체인 (ActiveDirectoryDefault)

ActiveDirectoryDefault 환경 변수, 워크로드 아이덴티티, 관리 신원 등 여러 자격 증명 제공자를 순서대로 시도합니다. 코드 변경 없이도 로컬 개발과 Azure 호스팅 서비스 모두에서 작동합니다.

인증에 관한 자세한 내용은 Microsoft Entra 인증을 참조하세요.

성능 팁

다음 기법들은 대량 복사 처리량을 극대화하는 데 도움을 줍니다.

대규모 데이터셋에는 생성기를 사용하세요

bulkcopy()는 생성기가 모든 이터러블을 받아들일 수 있기 때문에 메모리 사용을 최소화합니다:

def data_generator(count):
    """Generate rows without loading all into memory."""
    for i in range(count):
        yield (i, f"Item {i}", i * 1.5)

cursor = conn.cursor()
cursor.execute("""
    CREATE TABLE ##LargeDemo (ID INT, Name NVARCHAR(50), Value FLOAT)
""")
conn.commit()
result = cursor.bulkcopy("##LargeDemo", data_generator(1000))

더 빠른 로드를 위해 테이블 락을 사용하세요

동시 리더가 없을 때는 초기 로드가 많을 때 잠금 오버헤드를 줄이도록 설정 table_lock=True 하세요.

result = cursor.bulkcopy(
    "##LargeDemo",
    data,
    table_lock=True,
    batch_size=100000,
)

로드 시 인덱스를 비활성화하세요

대용량 로드 전에 비클러스터 인덱스를 일시적으로 비활성화하고, 이후 재구성하여 성능을 향상시키세요:

cursor = conn.cursor()

cursor.execute("""
    CREATE TABLE ##IndexDemo (ID INT, Name NVARCHAR(50), Value FLOAT)
""")
cursor.execute("CREATE NONCLUSTERED INDEX IX_Name ON ##IndexDemo(Name)")
conn.commit()

cursor.execute("ALTER INDEX IX_Name ON ##IndexDemo DISABLE")
conn.commit()

result = cursor.bulkcopy("##IndexDemo", data)
conn.commit()

cursor.execute("ALTER INDEX IX_Name ON ##IndexDemo REBUILD")
conn.commit()

테이블을 병렬로 로드

각 테이블마다 별도의 연결을 열고 부하를 동시에 실행하세요.

import concurrent.futures

def load_table(table_name, rows):
    conn = mssql_python.connect(connection_string)
    cursor = conn.cursor()
    cursor.execute(f"CREATE TABLE {table_name} (ID INT, Name NVARCHAR(50), Value FLOAT)")
    conn.commit()
    result = cursor.bulkcopy(table_name, rows)
    conn.commit()
    conn.close()
    return result["rows_copied"]

data = [(i, f"Item {i}", i * 1.5) for i in range(100)]

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    futures = [
        executor.submit(load_table, "##Load1", data),
        executor.submit(load_table, "##Load2", data),
        executor.submit(load_table, "##Load3", data),
    ]
    for future in concurrent.futures.as_completed(futures):
        print(f"Loaded {future.result()} rows")

대안과의 비교

다음 표는 대량 복사와 다른 데이터 삽입 방법을 비교합니다.

Method	사용 사례	성능
`cursor.bulkcopy()`	1,000행 이상의 대규모 데이터셋.	가장 빠름
`cursor.executemany()`	매개변수가 포함된 중간 데이터셋.	Moderate
`cursor.execute()` 루프 속에서	간단한 논리를 가진 작은 데이터셋.	가장 느린

피드백

이 페이지가 도움이 되었나요?

Last updated on 2026-07-22