mssql-python ile bir veri yükleme ve hareket deseni seçin

Sürücü, mssql-python Microsoft SQL'e veri yazmak için birden fazla yol sağlar. Her yol farklı iş yüklerine uyuyor. Bu rehber, veri hacminize, kaynak formatınıza ve güncel anlam sisteminize göre doğru olanı seçmenize yardımcı olur.

İş yüküne göre karar verin

İş yükü	Önerilen yol	Neden?
CSV dosyalarını bir tabloya yükleyin	CSV verilerini toplu kopya ile yükleyin	`bulkcopy()` Jeneratör ile herhangi bir boyuttaki dosyaları belleğe yüklemeden işliyor.
Uygulama kodundan tek bir satır ekle	Tek satırlı eklemeler	Düşük yük, basit hata işleme, üretilen anahtarları geri göndermek için OUTPUT ile çalışır.
Uygulama kodundan küçük-orta ölçekli bir parti ekleyin	Toplu eklemeler	Tek tek eklemelere kıyasla gidiş gelişleri azaltır.
Herhangi bir kaynaktan yüzlerce veya daha fazla satır yükleyin	Toplu kopya	TDS toplu ekleme, büyük hacimler için en verimli yoldur.
Bir anahtara göre satır ekleme veya güncelleme	Upsert ile MERGE	`MERGE`, INSERT, UPDATE ve DELETE öğelerini tek bir deyimde işler.
DataFrame'i tabloya yükle	Veri Çerçevelerini Yükle	pandas veya Polars'tan satırları alın ve `bulkcopy()` öğesine aktarın.
Parquet dosyaları aracılığıyla sahne verileri	Parquet hazırlama alanı	Ara dosya formatı gerektiğinde sistemler arası ETL için faydalıdır.

CSV verilerini toplu kopya ile yükleyin

CSV verilerini yüklemek, Python veritabanı çalışmaları için en yaygın sorudur. Bir jeneratör beslemesiyle csv.reader kullanın 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()

Üretici desen, dosya boyutu ne olursa olsun bellek kullanımını sabit tutar. Sütun eşlemesi ve kimlik işlemesi için bkz. Toplu kopyalama işlemleri.

Tek satırlı eklemeler

Her seferinde bir kaydı işlediğiniz uygulama düzeyindeki yazma işlemleri için tekli eklemeler kullanın. Oluşturulan anahtarları almak için OUTPUT INSERTED kullanın:

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

Tekli insertler şu durumlarda doğru seçimdir:

Her kullanıcı eylemi için bir satır eklersiniz (form gönderimi, API çağrısı).
Eklemeden önce her satırı tek tek doğrulamanız veya dönüştürmeniz gerekir.
Hemen eklenen ID veya diğer oluşturulan değerlere ihtiyacınız var.

Toplu eklemeler

Orta sayıda satırınız olduğunda ve toplu kopyalamanın sağladığı aktarım hızına ihtiyacınız olmadığında executemany() kullanın:

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() her satırı ayrı bir parametreli ifade olarak gönderir. Satır başına denetimden ziyade aktarım hızı daha önemli olduğunda, bulkcopy() TDS toplu ekleme protokolünü kullandığı için daha verimlidir. Geçiş noktası, satır genişliğine ve ağ gecikmesine bağlıdır; ancak genellikle birkaç yüz satırın alt sınırlarındadır.

Toplu kopya

İş hacmi satır bazında denetimden daha önemli olduğunda, bulkcopy() kullanın. Bu yöntem, satır satır eklemelerden çok daha verimli olan TDS toplu ekleme protokolünü kullanır:

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

Toplu kopyalama için performans ipuçları

Bellek kullanımını sabit tutmak için büyük veri setleri için üreteçler kullanın.
TDS toplu işlemi başına kaç satır gönderileceğini denetlemek için Set batch_size seçeneğini ayarlayın. 5.000 ile başlayın ve sıra genişliğine göre ayarlayın.
Özel yükler için masa kilitleri kullanın: cursor.bulkcopy("dbo.ProductImport", rows, table_lock=True).
Yüklemeden önce indeksleri devre dışı bırakın, sonra yeniden oluşturun. Bu dizi, yük sırasında indeks bakım yükünü önler.

Sütun eşlemeleri, kimlik sütunları, NULL işleme ve paralel yükleme için bkz. Toplu kopyalama işlemleri.

Upsert ile MERGE

MERGE, Microsoft SQL'de koşullu INSERT, UPDATE ve DELETE işlemlerini tek bir işlemde gerçekleştiren ifadedir. Python geliştiricilerinin sıkça ihtiyaç duyduğu "yeniyse ekle, varsa güncelle" desenini yönetiyor.

Tek satırlı upsert

Tek bir satır için, parametre takma adlarını tanımlayan bir USING yan tümcesiyle birlikte MERGE kullanın:

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

Sahneleme tablosu kullanarak toplu upsert

Toplu upsert işlemleri için, önce verileri geçici bir tabloya alın, ardından bu tablodan güncelleme yapmak için MERGE kullanın. DataFrame upsert işlemleri ve toplu güncellemeler için varsayılan kalıp olarak insert-or-update kullanın:

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

Bu örnek, varsayılan ekle-veya güncelle modelini göstermektedir:

INSERT Kaynakta bulunup hedefte bulunmayan satırlar (WHEN NOT MATCHED BY TARGET).
UPDATE her iki satırda da var olan satırlar (WHEN MATCHED).
OUTPUT maddesi, her satırda hangi işlemlerin yapıldığını bildirir ve bu da denetim izleri için faydalıdır.

Caution

WHEN NOT MATCHED BY SOURCE THEN DELETE öğesini yalnızca hazırlama verileri hedefin esas alınan tam bir kopyası olduğunda ekleyin. Eğer toplu işlem yalnızca değiştirilmiş satırları içeriyorsa, bu yan tümce kaynak akışından kasıtlı olarak çıkarılmış satırları siler.

Tam uzlaştırmaya ihtiyacınız varsa, kaynağın hedef tablo için yetkili olduğunu doğruladıktan sonra uzatın MERGE :

WHEN NOT MATCHED BY SOURCE THEN
    DELETE

Paylaşılan ortamlarda, eşzamanlı işler arasında çarpışmaları önlemek için her çalışma başına benzersiz bir küresel geçici tablo adı veya kalıcı bir aşamalama tablosu kullanın.

Bunun yerine ayrı UPDATE ve INSERT ifadelerinin ne zaman kullanılacağı

MERGE güçlüdür, ancak bazı uç durumlar içerir. Ayrı ifadeler kullanmayı düşünün:

Mantığa ihtiyacın DELETE yok. Ayrı bir UPDATE ve ardından INSERT WHERE NOT EXISTS kullanmak, daha okunaklıdır ve hata ayıklaması daha kolaydır.
Bu ifade MERGE o kadar karmaşık ki, kilitlenme davranışını tahmin etmek zor. Ayrı ifadeler, kilitleme ayrıntı düzeyi üzerinde doğrudan denetim sağlar.
MERGE kilit yükseltmesinin engellemeye neden olabileceği, yüksek eşzamanlılığa sahip bir tabloyu güncelliyorsunuz.

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

Veri Çerçevelerini Yükle

Pandas veya Polars DataFrame’inden satırları ayıklayın ve bulkcopy() kullanarak yükleyin:

pandas

Bir pandas DataFrame'i tuple'lara dönüştürün ve şu adrese bulkcopy()aktarın:

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

Kutuplar

Bir Polars DataFrame'i şu .rows() yöntemle tuple'lara dönüştürebilirsiniz:

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

Tam DataFrame yükleme desenleri için pandas entegrasyonu ve Polars entegrasyonuna bakınız.

Parquet hazırlama

Sistemler arasında veri taşırken veya ETL boru hattınız zaten Parquet dosyaları üretirken ara format olarak Parquet'i kullanın:

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

Büyük Parquet dosyaları için, bellek kullanımını sabit tutmak için satır grupları halinde okuyun:

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

Yüklü verileri doğrulama

Yüklendikten sonra sıra sayılarını doğrulayın ve verileri nokta kontrol edin:

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

Üretim yüklerinde, bulkcopy() çağrısını korumak için çağıran bağlantının işlemine güvenmeyin. bulkcopy() kendi dahili bağlantısını açar ve kopyalanan satırları bağımsız olarak onaylar; bu nedenle ana bağlantınızdaki conn.rollback() bunları geri alamaz. İki yaklaşım atomikliği sağlar:

Her partiyi kendi işlemine sarması için use_internal_transaction=True değerini ayarlayın. Yarı başarısız olan parti, o partiyi yarı yüklü bırakmak yerine geri çeker.
Verileri yükseltmeden önce doğrulamak için, verileri toplu olarak bir hazırlama tablosuna kopyalayın, doğrulayın ve ardından ana bağlantınızda bir işlem içinde INSERT ... SELECT kullanarak satırları hedef tabloya taşıyın. Bu, bağlantınız üzerinden çalıştığı için INSERT, doğrulama başarısız olursa conn.rollback() bunu geri alır.

# 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

Geri Bildirim

Bu sayfayı yararlı buldunuz mu?

Last updated on 2026-07-22