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

DataFrame bersifat kolom, jadi jalur tercepat adalah bulkcopy_arrow(), yang menggunakan tabel Arrow yang sudah diketahui panda cara memproduksinya. bulkcopy()mengambil row tuple, jadi Anda perlu meratakan kolom-kolom menjadi objek Python terlebih dahulu.

Konversikan tabel Arrow ke tipe data kolom tujuan sebelum Anda memuatnya. pyarrow menyimpulkan float64 untuk kolom numerik, yang tidak dapat dipetakan oleh driver ke uang, desimal, atau numerik:

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)

Tanpa cast, proses pemuatan gagal dengan error ValueError: Cannot map Arrow column 'Amount' (Float64) to SQL column 'Amount' (Money). Lakukan cast dengan Table.cast(), bukan meneruskan skema ke Table.from_pandas(), yang tidak dapat mengonversi kolom float langsung menjadi decimal128. NaN nilai menjadi SQL NULL pada jalur ini, jadi Anda tidak perlu menggantinya terlebih dahulu.

Jika Anda membutuhkan jalur row-tuple, itertuples() sudah menghasilkan tuple saat Anda melewati name=None:

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

Muat data Apache Arrow

Gunakan cursor.bulkcopy_arrow() untuk memuat data Apache Arrow. Metode ini membaca langsung dari memori Arrow, jadi Anda tidak perlu membangun tuple baris Python sebelum memanggilnya.

Argumen source menerima a pyarrow.Table, a pyarrow.RecordBatch, a pyarrow.RecordBatchReader, atau objek apa pun yang mengekspos antarmuka data Arrow C. Argumen yang tersisa sama dengan bulkcopy().

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

Setiap tipe kolom Arrow harus kompatibel dengan tipe kolom SQL tujuannya. Writer tidak mengonversi antar keluarga tipe data, jadi meneruskan kolom float64 ke kolom money akan memunculkan ValueError sebelum baris apa pun ditulis. Gunakan decimal128 untuk uang, desimal, dan kolom numerik .

Meneruskan sumber Arrow ke bulkcopy() memunculkan TypeError dan mengarahkan Anda ke bulkcopy_arrow().

Untuk informasi lebih lanjut tentang dukungan Arrow, termasuk cara melakukan streaming hasil dari satu tabel ke tabel lain, lihat integrasi Apache Arrow.

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. Berlaku untuk operasi salinan massal itu sendiri, bukan untuk koneksi internal. Gunakan `0` untuk menonaktifkan batas waktu operasi.
`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.

Note

bulkcopy() membuka koneksi internal terpisah ke server. Koneksi internal tersebut mewarisi batas waktu kueri kursor: atur Connection.timeout ke nilai positif sebelum Anda membuat kursor, dan nilai yang sama membatasi upaya koneksi penyalinan massal. Jika batas waktu kueri kursor adalah 0, koneksi internal menggunakan batas waktu koneksi bawaannya, yaitu 15 detik. Kursor mengambil nilai saat dibuat, jadi mengubah Connection.timeout setelahnya tidak memengaruhi kursor yang sudah ada atau salinan massal dalam penerbangan. Tingkatkan batas waktu kueri sebelum Anda membuat kursor untuk endpoint yang lambat, dibatasi lajunya, atau berlatensi tinggi (misalnya melalui VPN atau antarwilayah).

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.

Mulai dari sumber kolumnar

bulkcopy() mengambil iterable berisi tuple baris, jadi setiap nilai harus sudah tersedia sebagai objek Python sebelum proses penyalinan dimulai. Ketika data sudah dalam format kolumnar, bulkcopy_arrow() membaca buffer Arrow secara langsung dan melewati tahap itu. DataFrame pandas atau Polars, sebuah file Parquet, dan hasil dari cursor.arrow() semuanya merupakan sumber Arrow. Untuk informasi lebih lanjut, lihat Muat data Apache Arrow.

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_arrow()`	Dataset besar yang sudah bersifat kolom.	Tercepat
`cursor.bulkcopy()`	Dataset besar (lebih dari 1.000 baris) dari sumber berorientasi baris.	Cepat
`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-09-03