Use cópia em massa com mssql-python

O driver mssql-python inclui uma funcionalidade de cópia em massa que insere de forma eficiente grandes quantidades de dados no SQL Server, Base de Dados SQL do Azure, Azure SQL Managed Instance e base de dados SQL no Microsoft Fabric.

O cursor.bulkcopy() método fornece um caminho de alto desempenho para carregar grandes conjuntos de dados:

• Minimiza as viagens de ida e volta na rede.
• Opcionalmente, contorna a verificação de restrições durante a carga.
• Utiliza o protocolo TDS otimizado para inserção em massa.
• Alcança uma taxa de transferência comparável à de bcp.exe e SqlBulkCopy.

A extensão nativa baseada mssql_py_core em Rust alimenta a funcionalidade de cópia em massa. É executado fora do pipeline normal do cursor execute().

Utilização básica

Chame bulkcopy() num cursor, passando o nome da tabela de destino e um iterável de tuplos de linhas ou objetos Row:

Importante

Se criares ou alterares a tabela de destino na mesma sessão, chama conn.commit() antes de bulkcopy(). O protocolo de cópia em bloco utiliza um canal interno separado para ler os metadados da tabela, pelo que uma alteração DDL não confirmada pode causar um interbloqueio ou um tempo limite.

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

Valor de retorno

bulkcopy() retorna um dicionário:

Key	Tipo	Descrição
rows_copied	int	Número de linhas copiadas com sucesso.
batch_count	int	Número de lotes processados.
elapsed_time	float	Tempo necessário para a operação, em segundos.

Assinatura do método

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
)

Mapeamentos de coluna

Por defeito, bulkcopy() mapeia as colunas pela posição ordinal. Cada coluna de dados corresponde à coluna da tabela com o mesmo índice. Utilize o parâmetro column_mappings para substituir este comportamento.

Lista de nomes de colunas

Cada posição na lista corresponde ao índice de dados de origem:

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

Formato avançado: mapeamento explícito de índice

Cada tupla assume a forma (source_index, target_column_name). Use este formato para saltar ou reordenar as colunas:

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

Carregar a partir de ficheiros

Pode carregar dados a partir de ficheiros CSV e outros formatos de ficheiro passando um gerador para bulkcopy().

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

Ficheiros grandes com processamento em lotes

Defina o batch_size parâmetro para controlar quantas linhas o driver envia por lote. Esta abordagem funciona bem para ficheiros grandes:

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

Load pandas DataFrames

Converter um DataFrame do pandas numa lista de tuplas antes de o passar para 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)

Controlar valores NULL

Passe None em qualquer posição de coluna para inserir um valor 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)

Colunas de identidade

Para inserir valores identidade explícitos, defina 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)

Quando keep_identity=False (o predefinido), omite a coluna identidade dos teus dados e usa column_mappings para direcionar as colunas de não-identidade.

Opções de cópia em massa

Parâmetro	Default	Descrição
batch_size	0	Linhas por lote. 0 Deixa o servidor escolher o tamanho ótimo.
timeout	30	Tempo limite da operação em segundos.
keep_identity	False	Preservar os valores de identidade dos dados de origem.
check_constraints	False	Verifique as restrições da tabela durante o carregamento.
table_lock	False	Adquira um bloqueio ao nível da mesa em vez dos bloqueios ao nível da linha.
keep_nulls	False	Preservar os valores NULL em vez de inserir os valores predefinidos das colunas.
fire_triggers	False	Disparar INSERT gatilhos na mesa de alvo.
use_internal_transaction	False	Envolve cada lote numa transação interna.

Lidar com erros

bulkcopy() gera uma exceção se a carga falhar, por isso envolve a chamada num try/except bloco para detetar erros. Tem em mente que bulkcopy() é executado na sua própria ligação interna e confirma as linhas copiadas de forma independente, pelo que conn.rollback() na tua ligação principal não as pode anular. Para tornar um lote atómico, defina use_internal_transaction=True, que envolve cada lote numa transação própria que reverte automaticamente se o lote falhar:

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

Para sujeitar um carregamento à sua própria lógica de validação, copie em massa para uma tabela de preparação e, em seguida, transfira as linhas para a tabela de destino com um INSERT ... SELECT dentro de uma transação na sua conexão principal. Isso INSERT é executado na tua ligação, por isso conn.rollback() desfaz isso se a validação falhar.

Authentication

A cópia em massa utiliza um canal interno separado que requer o seu próprio token. O driver gere automaticamente a aquisição de tokens para os métodos de autenticação suportados.

Identidade gerida (ActiveDirectoryMSI)

Utilize Authentication=ActiveDirectoryMSI para identidade gerida atribuída pelo sistema ou pelo utilizador. Este método de autenticação é recomendado para serviços alojados no Azure, como VMs do Azure, App Service, Functions e 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")

Para uma identidade gerida atribuída pelo utilizador, passe o ID do cliente na cadeia de ligação:

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

Entidade de serviço (ActiveDirectoryServicePrincipal)

Use Authentication=ActiveDirectoryServicePrincipal para autenticação de principal de serviço (credenciais do cliente).

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

Cadeia de credenciais por defeito (ActiveDirectoryDefault)

ActiveDirectoryDefault Testa múltiplos fornecedores de credenciais em sequência, como variáveis de ambiente, identidade de carga de trabalho, identidade gerida e mais. Funciona tanto para desenvolvimento local como para serviços alojados no Azure sem alterações de código.

Para mais informações sobre autenticação, consulte autenticação Microsoft Entra.

Sugestões de desempenho

As técnicas seguintes ajudam-no a maximizar a taxa de transferência da cópia em massa.

Utilizar geradores para grandes conjuntos de dados

Os geradores minimizam o uso de memória porque bulkcopy() aceitam qualquer iterável:

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

Use fechaduras de mesa para cargas mais rápidas

Quando não tiveres leitores simultâneos, define table_lock=True para reduzir a sobrecarga associada ao bloqueio durante grandes carregamentos iniciais.

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

Desativar índices durante a carga

Desabilite temporariamente índices não agrupados antes da carga em massa e reconstrua-os depois para melhorar o desempenho:

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

Carregar tabelas em paralelo

Abrir uma ligação separada para cada tabela e executar as cargas em simultâneo.

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

Comparação com alternativas

A tabela seguinte compara a cópia em massa com outros métodos de inserção de dados.

Método	Caso de utilização	Desempenho
cursor.bulkcopy()	Grandes conjuntos de dados (mais de 1.000 linhas).	Mais rápido
cursor.executemany()	Conjuntos de dados médios com parâmetros.	Moderado
cursor.execute() num ciclo	Conjuntos de dados pequenos com lógica simples.	Mais lento

Conteúdo relacionado

• Execução de consultas
• Mapeamentos de tipo de dados
• Gestão de transações
• Afinação de desempenho