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

DataFrame sütun tabanlıdır, bu nedenle en hızlı yol, pandas’ın zaten üretebildiği Arrow tablosunu kullanan bulkcopy_arrow()’dir. bulkcopy() satır demetlerini alır, bu yüzden önce sütunları Python nesnelerine dönüştürmeniz gerekir.

Arrow tablosunu yüklemeden önce hedef sütun türlerine dönüştürün. pyarrow, sürücünün money, decimal veya float64 ile eşleyemediği sayısal bir sütun için çıkarımda bulunur:

import pandas as pd
import pyarrow as pa
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()

target = pa.schema([
    pa.field('ID', pa.int32()),
    pa.field('Name', pa.string()),
    pa.field('Amount', pa.decimal128(19, 4)),   # MONEY
])

table = pa.Table.from_pandas(df, preserve_index=False).cast(target)
result = cursor.bulkcopy_arrow("##PandasDemo", table)

Döküm olmadan yük başarısız olur ValueError: Cannot map Arrow column 'Amount' (Float64) to SQL column 'Amount' (Money). Şemayı Table.from_pandas() içine geçirmek yerine cast işlemini Table.cast() ile oluşturun; çünkü Table.from_pandas(), bir float sütununu doğrudan decimal128'ye dönüştüremez. NaN Bu yolda değerler SQL NULL olur, yani önce onları değiştirmenize gerek kalmaz.

Bunun yerine satır-tuple yoluna ihtiyacınız varsa, name=None geçirdiğinizde itertuples() zaten tuple döndürür:

data = list(df.itertuples(index=False, name=None))
result = cursor.bulkcopy("##PandasDemo", data)

Apache Arrow verilerini yükle

Apache Arrow verilerini yüklemek için cursor.bulkcopy_arrow() kullanın. Bu yöntem doğrudan Arrow belleğinden okur, bu yüzden Python satır tuple'larını çağırmadan önce oluşturmazsınız.

source bağımsız değişkeni, bir pyarrow.Table, bir pyarrow.RecordBatch, bir pyarrow.RecordBatchReader veya Arrow C veri arayüzünü sunan herhangi bir nesneyi kabul eder. Geri kalan argümanlar bulkcopy() ile aynıdır.

import mssql_python
import pyarrow as pa

conn = mssql_python.connect(connection_string)

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

cursor.execute("""
    CREATE TABLE ##ArrowDemo (ID INT, Name NVARCHAR(50), Amount FLOAT)
""")

table = pa.table({
    "ID": pa.array([1, 2, 3], type=pa.int32()),
    "Name": pa.array(["Alice", "Bob", "Carol"], type=pa.string()),
    "Amount": pa.array([50000.0, 60000.0, 55000.0], type=pa.float64()),
})

result = cursor.bulkcopy_arrow("##ArrowDemo", table)
print(f"Copied {result['rows_copied']} rows")

Her Ok sütun türü, hedef SQL sütun tipiyle uyumlu olmalıdır. Yazıcı, veri türü aileleri arasında dönüştürme yapmaz; bu nedenle bir float64 sütununun money sütununa geçirilmesi, herhangi bir satır yazılmadan önce ValueError hatasına neden olur. para birimi, ondalık ve decimal128 sütunlar için kullanın.

Bir Arrow kaynağını bulkcopy() içine geçirmek, TypeError hatasına neden olur ve sizi bulkcopy_arrow() konumuna yönlendirir.

Arrow desteği hakkında daha fazla bilgi için, bir sonuç setinin bir tablodan diğerine nasıl akış yapılacağı da dahil olmak üzere, Apache Arrow entegrasyonuna bakınız.

NULL değerleri işle

Bir SQL None değeri eklemek için NULL öğ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). Bu, toplu kopyalama işleminin kendisine uygulanır, dahili bağlantıya değil. Operasyon zaman aşımını devre dışı bırakmak için kullanın 0 .
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.

Note

bulkcopy() sunucuya ayrı bir iç bağlantı açar. Bu iç bağlantı, imlecin sorgu zamanlamasını devralır: imleci oluşturmadan önce pozitif bir değere ayarlanır Connection.timeout ve aynı değer toplu kopyalama bağlantı girişimini sınırlar. Eğer imlecin sorgu zaman aşımı 0ise, iç bağlantı varsayılan 15 saniyelik bağlantı zaman aşımını kullanır. Bir imleç, oluşturulduğu anda değerini alır; bu nedenle sonradan Connection.timeout değiştirilse bile mevcut bir imleç veya devam eden bir toplu kopyalama işlemi etkilenmez. Yavaş, kısıtlı veya yüksek gecikmeli uç noktalar için (örneğin VPN üzerinden veya bölgeler arasında) imleci oluşturmadan önce sorgu zamanını artırın.

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.

Sütunlu bir kaynaktan başlayın

bulkcopy()satır tuple'lardan oluşan bir iterable alır, bu yüzden her değerin kopya başlamadan önce bir Python nesnesi olarak var olması gerekir. Veri zaten sütunlu olduğunda, bulkcopy_arrow() Ok tamponlarını doğrudan okur ve o adımı atlar. Bir pandas veya Polars DataFrame, bir Parquet dosyası ve cursor.arrow() sonucunda elde edilen veri, Arrow kaynaklarıdır. Daha fazla bilgi için Load Apache Arrow verilerine bakınız.

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_arrow()	Zaten sütunlu olan büyük veri setleri.	En Hızlı
cursor.bulkcopy()	Satır odaklı kaynaklardan büyük veri setleri (1.000'den fazla satır).	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

• mssql-python ile sorguları çalıştır
• mssql-python için veri tipi eşlemeleri
• mssql-python ile işlem yönetimi
• mssql-python uygulamaları için performans ayarı