Choisissez un motif de chargement et de déplacement de données avec mssql-python

Le mssql-python pilote propose plusieurs voies pour écrire des données dans Microsoft SQL. Chaque parcours correspond à des charges de travail différentes. Ce guide vous aide à choisir le bon en fonction de votre volume de données, du format source et de la sémantique des mises à jour.

Décidez par charge de travail

Chargez les données CSV avec une copie en bloc

Le chargement de données CSV est la question d’ingestion la plus courante pour le travail sur une base de données Python. Utilisez csv.reader avec un générateur alimenté en 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()

Le motif générateur maintient l’utilisation de la mémoire constante quelle que soit la taille du fichier. Pour la correspondance des colonnes et la gestion des identités, voir Opérations de copie en masse.

Insertions sur une seule ligne

Utilisez des insertions unitaires pour les opérations d’écriture au niveau de l’application lorsque vous traitez un enregistrement à la fois. À utiliser OUTPUT INSERTED pour récupérer les clés générées :

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

Les inserts simples sont le bon choix lorsque :

• Vous insérez une ligne par action utilisateur (soumission de formulaire, appel API).
• Il faut valider ou transformer chaque ligne individuellement avant d’insérer.
• Vous avez besoin de l’ID inséré ou d’autres valeurs générées immédiatement.

Insertions par lots

Utilisez executemany() lorsque vous avez un nombre modéré de lignes et que vous n’avez pas besoin du débit de la copie en bloc :

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() envoie chaque ligne comme une instruction paramétrée distincte. Lorsque le débit importe plus que le contrôle ligne par ligne, bulkcopy() est plus efficace, car il utilise le protocole d’insertion en bloc TDS. Le point de bascule dépend de la largeur des lignes et de la latence du réseau, mais il se situe généralement de l’ordre de quelques centaines de lignes.

Copie par lots

Lorsque le débit compte plus que le contrôle par rangée, on utilise bulkcopy(). Il utilise le protocole d’insertion en bloc TDS, qui est nettement plus efficace que les insertions ligne par ligne :

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

Conseils de performance pour la copie en masse

• Utilisez des générateurs pour de grands ensembles de données afin de maintenir une utilisation mémoire constante.
• Définissez batch_size pour contrôler le nombre de lignes envoyées dans chaque lot TDS. Commence avec 5 000 et ajuste en fonction de la largeur des rangées.
• Utilisez des serrures de table pour des charges exclusives : cursor.bulkcopy("dbo.ProductImport", rows, table_lock=True).
• Désactivez les index avant de charger, puis reconstruisez après. Cette séquence évite la surcharge de maintenance de l’indice pendant la charge.

Pour les correspondances de colonnes, les colonnes d’identité, la gestion des valeurs NULL et le chargement parallèle, voir Opérations de copie en masse.

Mise à jour ou insertion avec MERGE

MERGEest l'instruction de Microsoft SQL pour les conditions INSERT, UPDATE, et DELETE dans une seule opération. Il gère le motif « insérer si nouveau, mettre à jour s’il existe » dont les développeurs Python ont souvent besoin.

Upsert à rangée unique

Pour une seule ligne, utilisez MERGE avec une USING clause définissant les alias des paramètres :

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

Upsert en vrac avec table de mise en scène

Pour les opérations d’upsert en masse, placez d’abord les données dans une table temporaire, puis utilisez MERGE pour effectuer la mise à jour à partir de celle-ci. Utilisez insert-or-update comme modèle par défaut pour les opérations d'upsert sur des DataFrame et les mises à jour par lot :

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

Cet exemple démontre le motif par défaut d’insertion ou de mise à jour :

• INSERT Lignes provenant de la source qui n’existent pas dans la cible (WHEN NOT MATCHED BY TARGET).
• UPDATE lignes présentes dans les deux (WHEN MATCHED).
• La clause OUTPUT rapporte les actions effectuées sur chaque ligne, ce qui est utile pour les traces d’audit.

Si vous avez besoin d’une réconciliation complète, n’étendez le MERGE qu’après avoir confirmé que la source fait autorité pour la table cible :

WHEN NOT MATCHED BY SOURCE THEN
    DELETE

Dans les environnements partagés, utilisez un nom unique global de table temporaire par exécution ou une table de staging permanente pour éviter les collisions entre tâches concurrentes.

Quand utiliser les instructions séparées UPDATE et INSERT à la place

MERGE est puissant, mais comporte des cas particuliers. Envisagez d’utiliser des instructions séparées lorsque :

• Tu n’as pas besoin DELETE de logique. Un UPDATE distinct suivi de INSERT WHERE NOT EXISTS est plus lisible et plus facile à déboguer.
• L’énoncé MERGE est suffisamment complexe pour que le comportement de verrouillage soit difficile à prévoir. Les instructions séparées vous donnent un contrôle explicite sur la granularité des verrous.
• Vous mettez à jour une table à forte concurrence où MERGE l’escalade de verrous pourrait provoquer des blocages.

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

Charger des DataFrames

Extraire des lignes d’un pandas ou d’un DataFrame Polars et les charger en utilisant bulkcopy():

pandas

Convertir un DataFrame pandas sous forme de tuples et le transmettre à bulkcopy() :

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

Polaires

Convertissez un DataFrame Polars en tuples en utilisant la .rows() méthode suivante :

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

Pour les modèles complets de chargement de DataFrame, voir l’intégration pandas et l’intégration Polars.

Scénographie en parquet

Utilisez Parquet comme format intermédiaire lors de la migration de données entre systèmes ou lorsque votre pipeline ETL produit déjà des fichiers Parquet :

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

Pour les grands fichiers Parquet, lisez en groupes de lignes pour maintenir une consommation mémoire constante :

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

Valider les données chargées

Après le chargement, vérifiez le nombre de lignes et faites des vérifications ponctuelles :

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

Pour les charges de production, ne comptez pas sur la transaction de la connexion appelante pour protéger un bulkcopy() appel. bulkcopy() ouvre sa propre connexion interne et valide les lignes copiées indépendamment, donc un conn.rollback() sur votre connexion principale ne peut pas annuler leur validation. Deux approches garantissent l’atomicité :

• Configurez use_internal_transaction=True pour emballer chaque lot dans sa propre transaction. Un lot qui échoue en cours de traitement annule ce lot au lieu de le laisser partiellement chargé.
• Pour valider les données avant de les transférer vers la table cible, copiez-les en bloc dans une table intermédiaire, validez-les, puis déplacez les lignes vers la table cible en utilisant un INSERT ... SELECT dans le cadre d’une transaction sur la connexion principale. Étant donné que cela INSERT s’exécute sur votre connexion, conn.rollback() l’annule si la validation échoue.

# 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

Contenu connexe

• Opérations de copie en bloc
• intégration de pandas
• Intégration polaire
• Réglage des performances
• Gestion des transactions
• Accès aux données et modes d’analyse