Utilisez la copie en vrac avec mssql-python

Le pilote mssql-python inclut une fonction de copie en masse qui insère efficacement de grandes quantités de données dans SQL Server, Azure SQL Database, Azure SQL Managed Instance et la base de données SQL dans Microsoft Fabric.

La cursor.bulkcopy() méthode offre un chemin haute performance pour charger de grands ensembles de données :

• Minimise les allers-retours du réseau.
• Éventuellement, elle contourne la vérification des contraintes pendant la charge.
• Utilise le protocole TDS optimisé pour l’insertion de données en masse.
• Atteint un débit comparable à bcp.exe et SqlBulkCopy.

L’extension native basée mssql_py_core sur Rust alimente la fonction de copie en masse. Il s’exécute en dehors du pipeline normal du curseur execute().

Utilisation de base

Appelez bulkcopy() sur un curseur, en passant le nom de la table cible et un itérable de tuples de lignes ou d’objets Row :

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

Valeur renvoyée

bulkcopy() renvoie un dictionnaire :

Signature de méthode

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
)

Mappages de colonnes

Par défaut, bulkcopy() associe les colonnes en fonction de leur position ordinale. Chaque colonne de données correspond à la colonne du tableau au même index. Utilisez ce column_mappings paramètre pour contourner ce comportement.

Liste des noms de colonnes

Chaque position dans la liste correspond à l’index des données sources :

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

Format avancé : cartographie explicite des indices

Chaque tuple prend la forme (source_index, target_column_name). Utilisez ce format pour sauter ou réorganiser les colonnes :

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

Chargement à partir de fichiers

Vous pouvez charger des données à partir de fichiers CSV et d’autres formats de fichiers en passant un générateur vers bulkcopy().

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

Grands fichiers avec batching

Réglez le batch_size paramètre pour contrôler le nombre de lignes que le pilote envoie par lot. Cette approche fonctionne bien pour les fichiers volumineux :

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

Charger les pandas DataFrames

Un DataFrame a une structure colonnaire, donc la voie la plus rapide est bulkcopy_arrow(), qui consomme la table Arrow que pandas sait déjà produire. bulkcopy() prend des tuples de lignes, donc il faut d’abord convertir les colonnes en objets Python simples.

Convertissez la table Arrow en types des colonnes de destination avant de la charger. pyarrow déduit float64 pour une colonne numérique, que le conducteur ne peut pas associer à de la monnaie, décimale ou numérique :

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)

Sans le cast, la charge échoue avec ValueError: Cannot map Arrow column 'Amount' (Float64) to SQL column 'Amount' (Money). Construis la distribution avec Table.cast() plutôt que de passer le schéma à Table.from_pandas(), qui ne peut pas convertir directement une colonne flottante decimal128 . NaN les valeurs deviennent SQL NULL sur ce chemin, donc vous n’avez pas besoin de les remplacer d’abord.

Si vous avez besoin du chemin de tuple de ligne à la place, itertuples() renvoie déjà des tuples si vous lui passez name=None :

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

Charger les données Apache Arrow

Utilisez cursor.bulkcopy_arrow() pour charger les données Apache Arrow. Cette méthode lit directement depuis la mémoire Arrow, donc vous ne construisez pas de tuples de lignes Python avant de l'appeler.

L’argument source accepte un pyarrow.Table, un pyarrow.RecordBatch, un pyarrow.RecordBatchReader, ou tout objet qui expose l’interface de données Arrow C. Les autres arguments sont les mêmes que 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")

Chaque type de colonne Arrow doit être compatible avec son type de colonne SQL de destination. L’auteur ne convertit pas entre les familles de types, donc passer une float64 colonne à une colonne d’argent augmente ValueError avant que les lignes ne soient écrites. Utilisez decimal128 pour l’argent, les décimales et les colonnes numériques .

Passer une source de flèche vers bulkcopy() augmente TypeError et vous dirige vers bulkcopy_arrow().

Pour plus d’informations sur le support d’Arrow, y compris comment diffuser un ensemble de résultats d’une table vers une autre, voir intégration Apache Arrow.

Gérer les valeurs NULL

Passez None dans n’importe quelle position de colonne pour insérer une valeur SQL NULL :

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)

Colonnes d’identité

Pour insérer des valeurs d’identité explicites, on fixe 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)

Lorsque keep_identity=False (par défaut), omettez la colonne d’identité de vos données et utilisez column_mappings pour cibler les colonnes autres que la colonne d’identité.

Options de copies en vrac

Gérer les erreurs

bulkcopy() crée une exception si le chargement échoue, donc encapsulez l’appel dans un try/except bloc pour détecter les erreurs. Gardez à l’esprit que bulkcopy() utilise sa propre connexion interne et valide les lignes copiées de manière indépendante ; un conn.rollback() sur votre connexion principale ne peut donc pas les annuler. Pour rendre un lot atomique, on définit use_internal_transaction=True, qui enveloppe chaque lot dans sa propre transaction qui revient automatiquement en cas d’échec du lot :

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

Pour faire passer un chargement derrière votre propre logique de validation, copiez les données en bloc dans une table intermédiaire, puis promouvez les lignes vers la table cible avec un INSERT ... SELECT à l’intérieur d’une transaction via votre connexion principale. Cela INSERT s’exécute via votre connexion, donc conn.rollback() l’annule si la validation échoue.

Authentication

La copie en bloc utilise un canal interne séparé qui nécessite son propre jeton. Le pilote gère automatiquement l’acquisition des jetons pour les méthodes d’authentification prises en charge.

Identité gérée (ActiveDirectoryMSI)

À utiliser Authentication=ActiveDirectoryMSI pour une identité managée assignée par le système ou par l’utilisateur. Cette méthode d’authentification est recommandée pour les services hébergés sur Azure tels que les machines virtuelles Azure, App Service, Functions et 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")

Pour une identité managée attribuée par l’utilisateur, passez l’ID client dans la chaîne de connexion :

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

Principal du service (ActiveDirectoryServicePrincipal)

Utilisation Authentication=ActiveDirectoryServicePrincipal pour l’authentification du principal de service (identifiants clients).

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

Chaîne d’identifiants par défaut (ActiveDirectoryDefault)

ActiveDirectoryDefault essaie successivement plusieurs fournisseurs d’informations d’identification, tels que les variables d’environnement, l’identité de charge de travail, l’identité gérée, entre autres. Il fonctionne aussi bien pour le développement local que pour les services hébergés sur Azure sans modification de code.

Pour plus d’informations sur l’authentification, voir authentification Microsoft Entra.

Astuces pour les performances

Les techniques suivantes vous aident à maximiser le débit de copies en masse.

Partez d’une source en colonne

bulkcopy()prend un itérable de tuples de lignes, donc chaque valeur doit exister sous forme d'objet Python avant que la copie ne commence. Lorsque les données sont déjà colonnaires, bulkcopy_arrow() lit directement les tampons Arrow et ignore cette étape. Un pandas ou Polars DataFrame, un fichier Parquet, et le résultat de cursor.arrow() sont tous des sources Arrow. Pour plus d’informations, voir Charger les données Apache Arrow.

Utilisez des générateurs pour de grands ensembles de données

Les générateurs minimisent l’utilisation de la mémoire car bulkcopy() acceptent tout itérable :

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

Utilisez des serrures de table pour des charges plus rapides

Lorsque vous n’avez pas de lecteurs simultanés, définissez table_lock=True afin de réduire la surcharge liée au verrouillage lors de chargements initiaux volumineux.

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

Désactiver les index pendant le chargement

Désactivez temporairement les index non clusterisés avant la charge en masse et reconstruisez-les ensuite pour améliorer les performances :

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

Charger les tables en parallèle

Ouvrez une connexion séparée pour chaque table et exécutez les charges simultanément.

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

Comparaison avec les alternatives

Le tableau suivant compare la copie en vrac avec d’autres méthodes d’insertion de données.

Contenu connexe

• Exécuter des requêtes avec mssql-python
• Mappage de types de données pour mssql-python
• Gestion des transactions avec mssql-python
• Réglage des performances pour les applications mssql-python