Menggunakan salinan massal dengan mssql-python

Driver mssql-python menyertakan fitur salinan massal yang secara efisien menyisipkan data dalam jumlah besar ke SQL Server, Azure SQL Database, Azure SQL Managed Instance, dan database SQL di Microsoft Fabric.

Metode ini cursor.bulkcopy() menyediakan jalur performa tinggi untuk memuat himpunan data besar:

Meminimalkan komunikasi bolak-balik melalui jaringan.
Secara opsional melewati pemeriksaan batasan selama pemuatan.
Menggunakan protokol penyisipan massal TDS yang dioptimalkan.
Mencapai throughput yang sebanding dengan bcp.exe dan SqlBulkCopy.

Ekstensi asli berbasis mssql_py_core Rust mendukung fitur penyalinan massal. Ini beroperasi di luar alur kursor normal execute().

Penggunaan dasar

Panggil bulkcopy() pada kursor, dengan meneruskan nama tabel target dan iterable berisi tuple baris atau objek Row:

Penting

Jika Anda membuat atau mengubah tabel target dalam sesi yang sama, panggil conn.commit() sebelum bulkcopy(). Protokol penyalinan massal menggunakan saluran internal terpisah untuk membaca metadata tabel, sehingga perubahan DDL yang tidak diterapkan dapat menyebabkan kebuntuan atau batas waktu.

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

Mengembalikan nilai

bulkcopy() mengembalikan kamus:

Tombol	Jenis	Deskripsi
`rows_copied`	int	Jumlah baris yang berhasil disalin.
`batch_count`	int	Jumlah batch yang diproses.
`elapsed_time`	float	Waktu yang dibutuhkan untuk operasi dalam hitungan detik.

Tanda tangan metode

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
)

Pemetaan kolom

Secara default, bulkcopy() memetakan kolom berdasarkan posisi ordinal. Setiap kolom data dipetakan ke kolom tabel dengan indeks yang sama. Gunakan parameter column_mappings untuk mengganti perilaku ini.

Daftar nama kolom

Setiap posisi dalam daftar sesuai dengan indeks data sumber:

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

Format lanjutan: pemetaan indeks eksplisit

Setiap tuple mengambil bentuk (source_index, target_column_name). Gunakan format ini untuk melewati atau menyusun ulang kolom:

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

Muat dari file

Anda dapat memuat data dari file CSV dan format file lainnya dengan meneruskan generator ke bulkcopy().

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

File berukuran besar dengan pemrosesan batch

Atur batch_size parameter untuk mengontrol berapa banyak baris yang dikirim driver per batch. Pendekatan ini bekerja dengan baik untuk file besar:

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

Muat DataFrame pandas

Konversi DataFrame panda ke daftar tuple sebelum meneruskannya ke bulkcopy():

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)

Menangani nilai NULL

Masukkan None pada posisi kolom mana pun untuk menyisipkan nilai NULL SQL:

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)

kolom identitas

Untuk menyisipkan nilai identitas eksplisit, atur 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)

Ketika keep_identity=False (default), hilangkan kolom identitas dari data Anda dan gunakan column_mappings untuk menargetkan kolom nonidentitas.

Opsi salinan massal

Parameter	Default	Deskripsi
`batch_size`	`0`	Baris per kelompok. `0` memungkinkan server memilih ukuran optimal.
`timeout`	`30`	Batas waktu operasi dalam hitungan detik.
`keep_identity`	`False`	Pertahankan nilai identitas dari data sumber.
`check_constraints`	`False`	Periksa batasan tabel selama pemuatan.
`table_lock`	`False`	Dapatkan kunci tingkat tabel alih-alih kunci tingkat baris.
`keep_nulls`	`False`	Pertahankan nilai NULL alih-alih menyisipkan default kolom.
`fire_triggers`	`False`	Picu trigger INSERT pada tabel target.
`use_internal_transaction`	`False`	Bungkus setiap batch dalam transaksi internal.

Menangani kesalahan

bulkcopy() memunculkan pengecualian jika beban gagal, jadi bungkus panggilan dalam try/except blok untuk menangkap kesalahan. Perlu diingat bahwa bulkcopy() berjalan pada koneksi internalnya sendiri dan meng-commit baris yang disalin secara independen, sehingga conn.rollback() pada koneksi utama Anda tidak dapat membatalkan perubahan tersebut. Untuk menjadikan batch bersifat atomik, atur use_internal_transaction=True, yang membungkus tiap batch dalam transaksinya sendiri yang dibatalkan secara otomatis jika batch gagal:

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

Untuk membuat pemuatan data bergantung pada logika validasi Anda sendiri, lakukan salin massal ke tabel antara, lalu pindahkan baris-baris tersebut ke tabel target dengan INSERT ... SELECT di dalam transaksi pada koneksi utama Anda. Hal itu INSERT berjalan melalui koneksi Anda, jadi conn.rollback() membatalkannya jika validasi gagal.

Authentication

Salinan massal menggunakan saluran internal terpisah yang memerlukan tokennya sendiri. Driver menangani akuisisi token secara otomatis untuk metode autentikasi yang didukung.

Identitas terkelola (ActiveDirectoryMSI)

Gunakan Authentication=ActiveDirectoryMSI untuk identitas terkelola yang ditetapkan sistem atau ditetapkan pengguna. Metode autentikasi ini direkomendasikan untuk layanan yang dihosting Azure seperti VM Azure, App Service, Functions, dan AKS.

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

Untuk identitas terkelola yang ditetapkan pengguna, teruskan ID klien dalam string koneksi:

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

Perwakilan layanan (ActiveDirectoryServicePrincipal)

Gunakan Authentication=ActiveDirectoryServicePrincipal untuk autentikasi perwakilan layanan (kredensial klien).

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

Rantai kredensial default (ActiveDirectoryDefault)

ActiveDirectoryDefault mencoba beberapa penyedia kredensial secara berurutan, seperti variabel lingkungan, identitas beban kerja, identitas terkelola, dan banyak lagi. Ini berfungsi untuk pengembangan lokal dan layanan yang dihosting Azure tanpa perubahan kode.

Untuk informasi selengkapnya tentang autentikasi, lihat autentikasi Microsoft Entra.

Tips kinerja

Teknik berikut membantu Anda memaksimalkan throughput salinan massal.

Menggunakan generator untuk himpunan data besar

Generator meminimalkan penggunaan memori karena bulkcopy() dapat menerima objek iterable apa pun:

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

Gunakan kunci meja untuk pemuatan yang lebih cepat

Jika Anda tidak memiliki pembaca simultan, atur table_lock=True untuk mengurangi beban penguncian selama pemuatan awal dalam jumlah besar.

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

Menonaktifkan indeks selama pemuatan

Nonaktifkan sementara indeks nonkluster sebelum pemuatan data massal dan bangun ulang indeks tersebut setelahnya untuk meningkatkan kinerja:

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

Muat tabel secara paralel

Buka koneksi terpisah untuk setiap tabel dan jalankan beban secara bersamaan.

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

Perbandingan dengan alternatif

Tabel berikut membandingkan salinan massal dengan metode penyisipan data lainnya.

Metode	Skenario penggunaan	Kinerja
`cursor.bulkcopy()`	Himpunan data besar (lebih dari 1.000 baris).	Tercepat
`cursor.executemany()`	Kumpulan data berukuran sedang dengan parameter.	Sedang
`cursor.execute()` dalam lingkaran	Himpunan data kecil dengan logika langsung.	Paling lambat

Saran dan Komentar

Apakah halaman ini membantu?

Last updated on 2026-07-22