mssql-python ile toplu kopyalama kullanın

mssql-python sürücüsü, SQL Server, Azure SQL Veritabanı, Azure SQL Yönetilen Örneği ve Microsoft Fabric'teki SQL veritabanına büyük miktarda veriyi verimli şekilde ekleyen toplu kopyalama özelliği içerir.

Yöntem, cursor.bulkcopy() büyük veri setlerinin yüklenmesi için yüksek performanslı bir yol sağlar:

• Ağ gidiş-dönüş mesafesini en aza indirir.
• Yük sırasında kısıtlama kontrolü isteğe bağlı olarak atlar.
• Optimize edilmiş TDS toplu insert protokolünü kullanır.
• bcp.exe ve SqlBulkCopy ile karşılaştırılabilir iş hacmi sağlar.

Rust tabanlı mssql_py_core yerel uzantı toplu kopyalama özelliğini güçlendiriyor. Normal imleç execute() işlem hattının dışında çalışır.

Temel kullanım

Bir imleç üzerinde bulkcopy() çağrısı yapın; hedef tablo adını ve satır demetlerinden veya Row nesnelerinden oluşan yinelenebilir bir öğeyi geçirin:

Important

Hedef tabloyu aynı oturumda oluşturur veya değiştirirseniz, conn.commit() öğesini bulkcopy() öğesinden önce çağırın. Toplu kopyalama protokolü, tablo meta verilerini okumak için ayrı bir dahili kanal kullanır; bu nedenle henüz işlenmemiş bir DDL değişikliği kilitlenmeye veya zaman aşımına neden olabilir.

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']}")

Dönüş değeri

bulkcopy() bir sözlük döndürür:

Key	Türü	Açıklama
rows_copied	int	Başarıyla kopyalanan satır sayısı.
batch_count	int	İşlenen parti sayısı.
elapsed_time	float	Operasyon için süren süre saniyelerde.

Yöntem imzası

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
)

Sütun eşlemeleri

Varsayılan olarak, bulkcopy() sütunları sırasal konuma göre eşler. Her veri sütunu aynı indeksteki tablo sütununa eşlenir. Bu davranışı geçersiz kılmak için parametreyi column_mappings kullanın.

Sütun isim listesi

Listedeki her pozisyon, kaynak veri indeksine karşılık gelir:

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

Gelişmiş format: açık indeks eşlemesi

Her tuple (source_index, target_column_name) biçimini alır. Bu formatı kullanarak sütunları atlamak veya yeniden sıralamak için:

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

Dosyalardan yükleme

bulkcopy() öğesine bir üreteç geçirerek CSV dosyalarından ve diğer dosya biçimlerinden veri yükleyebilirsiniz.

CSV dosyası

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

Toplu işlemeli büyük dosyalar

Parametreyi batch_size , sürücünün bir parti başına kaç satır göndereceğini kontrol edecek şekilde ayarlayın. Bu yaklaşım büyük dosyalar için iyi çalışır:

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 DataFrames Yükle

Bir pandas DataFrame'i bir tuples listesine dönüştürün, sonra onu şu adrese bulkcopy()aktarın:

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 değerleri işle

Bir SQL NULL değeri eklemek için None öğesini herhangi bir sütun konumuna geçirin:

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)

Kimlik sütunları

Açıkça belirtilmiş identity değerlerini eklemek için keep_identity=True değerini ayarlayın:

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 (varsayılan) olduğunda, verilerinizden kimlik sütununu çıkarın ve kimlik dışındaki sütunları hedeflemek için column_mappings kullanın.

Toplu kopya seçenekleri

Parametre	Varsayılan	Açıklama
batch_size	0	Toplu işlem başına satır sayısı. 0 sunucunun optimal boyutu seçmesini sağlar.
timeout	30	İşlem zaman aşımı (saniye cinsinden).
keep_identity	False	Kaynak verilerden kimlik değerlerini koruyun.
check_constraints	False	Yükleme sırasında tablo kısıtlamalarını kontrol edin.
table_lock	False	Sıra seviyesindeki kilitler yerine masa seviyesinde bir kilit edinin.
keep_nulls	False	Sütun için varsayılan değerler eklemek yerine NULL değerlerini koruyun.
fire_triggers	False	Hedef tabloda INSERT tetikleyicileri tetikleyin.
use_internal_transaction	False	Her partiyi dahili bir işlemle paketleyin.

Hatalarla başa çıkma

bulkcopy() Yük başarısız olursa istisna çıkarır, bu yüzden hataları yakalamak için çağrıyı bir try/except blok içine sarar. bulkcopy() öğesinin kendi dahili bağlantısı üzerinden çalıştığını ve kopyalanan satırları bağımsız olarak commit ettiğini unutmayın; bu nedenle ana bağlantınızdaki bir conn.rollback() bunları geri alamaz. Bir toplu işlemi atomik hâle getirmek için, her toplu işlemi başarısız olması durumunda otomatik olarak geri alınan kendi işlemi içinde saran use_internal_transaction=True ayarını yapın:

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}")

Bir yükleme işlemini kendi doğrulama mantığınızın arkasına almak için, verileri önce bir hazırlama tablosuna toplu olarak kopyalayın, ardından ana bağlantınızda bir işlem içinde INSERT ... SELECT kullanarak satırları hedef tabloya aktarın. Bu INSERT bağlantınızda çalışır, bu nedenle doğrulama başarısız olursa conn.rollback() bunu geri alır.

Authentication

Toplu kopyalama, kendi tokenı gerektiren ayrı bir dahili kanal kullanır. Sürücü, desteklenen kimlik doğrulama yöntemleri için token edinimini otomatik olarak yönetir.

Yönetilen kimlik (ActiveDirectoryMSI)

Sistem tarafından atanan veya kullanıcı tarafından atanan yönetilen kimlik için Authentication=ActiveDirectoryMSI kullanın. Bu kimlik doğrulama yöntemi, Azure VM'leri, App Service, Functions ve AKS gibi Azure barındırılan hizmetler için önerilir.

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

Kullanıcı tarafından atanan yönetilen bir kimlik için, istemci kimliğini bağlantı dizesi'e aktarın:

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

Hizmet sorumlusu (ActiveDirectoryServicePrincipal)

Hizmet sorumlusu (istemci kimlik bilgileri) kimlik doğrulaması için Authentication=ActiveDirectoryServicePrincipal kullanın.

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

Varsayılan kimlik doğrulama zinciri (ActiveDirectoryDefault)

ActiveDirectoryDefault Ortam değişkenleri, iş yükü kimliği, yönetilen kimlik ve daha fazlası gibi birden fazla kimlik kaynağı sağlayıcısını sırayla dener. Hem yerel geliştirme hem de Azure barındırılan hizmetler için kod değişikliği olmadan çalışır.

Kimlik doğrulama hakkında daha fazla bilgi için Microsoft Entra doğrulama bölümüne bakınız.

Performans ipuçları

Aşağıdaki teknikler, toplu kopya verimliliğini en üst düzeye çıkarmanıza yardımcı olur.

Büyük veri kümeleri için üreteçler kullanın

Oluşturucular, her türlü iterable kabul ettiği için bellek kullanımını bulkcopy() en aza indirir:

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

Daha hızlı yüklemeler için masa kilitleri kullanın

Eşzamanlı okuyucunuz yoksa, büyük ilk veri yüklemeleri sırasında kilitleme ek yükünü azaltmak için table_lock=True’yi ayarlayın.

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

Yükleme sırasında indeksleri devre dışı bırakın

Toplu yüklemeden önce kümelenmiş olmayan indeksleri geçici olarak devre dışı bırakın ve performansı artırmak için sonrasında yeniden oluşturun:

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

Tabloları paralel olarak yükleyin

Her tablo için ayrı bir bağlantı açın ve yükleri eşzamanlı çalıştırın.

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

Alternatiflerle karşılaştırma

Aşağıdaki tablo, toplu kopyalamaları diğer veri ekleme yöntemleriyle karşılaştırmaktadır.

Method	Kullanım örneği	Performans
cursor.bulkcopy()	Büyük veri setleri (1.000'den fazla satır).	En Hızlı
cursor.executemany()	Parametreli orta veri setleri.	Orta
cursor.execute() bir döngü içinde	Basit mantıklı küçük veri setleri.	En yavaş

İlgili içerik

• Sorguların yürütülmesi
• Veri türü eşlemeleri
• İşlem yönetimi
• Performans ayarlama