Escolha um padrão de carregamento e movimento de dados com mssql-python

O mssql-python driver oferece múltiplos caminhos para gravar dados no Microsoft SQL. Cada caminho se encaixa em diferentes cargas de trabalho. Este guia ajuda você a escolher a opção certa com base no volume de dados, formato de origem e semântica de atualização.

Decida com base na carga de trabalho

Carregar dados CSV com cópia em massa

Carregar dados CSV é a pergunta de ingestimento mais comum para trabalhos com bancos de dados em Python. Use csv.reader com um gerador que alimenta 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()

O padrão gerador mantém o uso de memória constante independentemente do tamanho do arquivo. Para mapeamento de colunas e tratamento de identidade, veja Operações de cópia em massa.

Inserções de uma única fileira

Use inserções individuais para operações de gravação no nível da aplicação, quando você processa um registro por vez. Use OUTPUT INSERTED para recuperar chaves geradas:

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

Insertos individuais são a escolha certa quando:

• Você insere uma linha por ação do usuário (envio de formulário, chamada de API).
• Você precisa validar ou transformar cada linha individualmente antes de inserir.
• Você precisa do ID inserido ou outros valores gerados imediatamente.

Inserções em lote

Use executemany() quando você tiver um número moderado de linhas e não precisar da taxa de transferência da cópia em massa:

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() envia cada linha como uma instrução parametrizada separada. Quando a taxa de transferência importa mais do que o controle por linha, bulkcopy() é mais eficiente porque utiliza o protocolo TDS de inserção em massa. O ponto de transição depende da largura de cada linha e da latência da rede, mas geralmente fica na faixa de algumas centenas de linhas.

Cópia em lote

Quando a taxa de transferência importa mais do que o controle por linha, use bulkcopy(). Ele utiliza o protocolo TDS de inserção em massa, que é significativamente mais eficiente do que inserções linha a linha:

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

Dicas de desempenho para cópia em massa

• Use geradores para conjuntos de dados grandes e mantenha o consumo de memória constante.
• Defina batch_size para controlar quantas linhas são enviadas em cada lote TDS. Comece com 5.000 e ajuste de acordo com a largura da linha.
• Use bloqueios de tabela para cargas exclusivas: cursor.bulkcopy("dbo.ProductImport", rows, table_lock=True).
• Desative os índices antes do carregamento e reconstrua-os depois. Essa sequência evita a sobrecarga de manutenção do índice durante a carga.

Para mapeamentos de colunas, colunas identidade, tratamento de NULL e carregamento paralelo, consulte Operações de cópia em massa.

Upsert com MERGE

MERGE é a instrução do Microsoft SQL para INSERT, UPDATE e DELETE condicionais em uma única operação. Ele lida com o padrão "inserir se for novo, atualizar se existir" que os desenvolvedores Python normalmente precisam.

Upsert de fileira única

Para uma única linha, use MERGE com uma USING cláusula que defina aliases de parâmetros:

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 em massa com uma mesa de preparação

Para upserts em massa, coloque os dados em uma tabela temporária primeiro e depois use MERGE para atualizar a partir dela. Use insert-or-update como padrão para operações de upsert em DataFrames e atualizações em lote:

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

Este exemplo demonstra o padrão padrão de inserção ou atualização:

• INSERT linhas da fonte que não existem no alvo (WHEN NOT MATCHED BY TARGET).
• UPDATE linhas presentes em ambos (WHEN MATCHED).
• A cláusula OUTPUT informa qual ação foi tomada em cada linha, o que é útil para trilhas de auditoria.

Se precisar de reconciliação completa, estenda a MERGE somente depois de confirmar que a fonte é a fonte confiável da tabela de destino:

WHEN NOT MATCHED BY SOURCE THEN
    DELETE

Em ambientes compartilhados, use um nome único de tabela temporária global por execução ou uma tabela permanente de staging para evitar colisões entre trabalhos concorrentes.

Quando usar instruções separadas UPDATE e INSERT em vez disso

MERGE é poderoso, mas tem casos excepcionais. Considere usar instruções separadas quando:

• Você não precisa de lógica DELETE. Um UPDATE separado, seguido de INSERT WHERE NOT EXISTS, é mais legível e mais simples de depurar.
• A MERGE instrução é complexa o suficiente para que o comportamento de bloqueio seja difícil de prever. Declarações separadas permitem controle explícito sobre a granularidade do bloqueio.
• Você está atualizando uma tabela de alta concorrência em que o MERGE escalonamento de bloqueios pode causar bloqueios.

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

Carregar DataFrames

Extraia linhas de um pandas ou Polars DataFrame e carregue-as usando bulkcopy():

pandas

Converta um DataFrame do pandas para tuplas e passe-o para 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()

Polars

Converta um DataFrame do Polars em tuplas usando o método .rows():

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

Para padrões completos de carregamento do DataFrame, veja integração com pandas e integração com Polars.

Área de preparação do Parquet

Use o Parquet como formato intermediário ao migrar dados entre sistemas ou quando seu pipeline ETL já produz arquivos 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()

Para arquivos grandes de Parquet, leia em grupos de linhas para manter o uso de memória 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()

Validar dados carregados

Após o carregamento, verifique a contagem de linhas e verifique os dados pontuais:

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

Para cargas de trabalho de produção, não confie na transação da conexão chamadora para proteger uma chamada bulkcopy(). bulkcopy() abre uma conexão interna própria e faz commit das linhas copiadas independentemente, então um conn.rollback() na conexão principal não pode desfazê-las. Duas abordagens garantem atomicidade:

• Configure use_internal_transaction=True para envolver cada lote em sua própria transação. Um lote que falha no meio do caminho reverte esse lote em vez de deixá-lo meio carregado.
• Para validar os dados antes de promovê-los, copie-os em massa para uma tabela de preparação, valide-os e, em seguida, mova as linhas para a tabela de destino usando um(a) INSERT ... SELECT dentro de uma transação na conexão principal. Como isso INSERT é executado na sua conexão, conn.rollback() o desfaz caso a validação falhe.

# 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

Conteúdo relacionado

• Operações de cópia em massa
• integração do Pandas
• Integração com Polars
• Otimização do desempenho
• Gerenciamento de transações
• Padrões de acesso a dados e análises