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 n-uplet 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

Convertissez un DataFrame pandas en une liste de tuples avant de la transmettre à 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)

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

Service principal (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.

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écution des requêtes
• Mappages de types de données
• Gestion des transactions
• Réglage des performances