Use cópia em massa com mssql-python

O driver mssql-python inclui um recurso de cópia em massa que insere eficientemente grandes volumes de dados no SQL Server, Banco de Dados SQL do Azure, Instância Gerenciada de SQL do Azure e banco de dados SQL no Microsoft Fabric.

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

• Minimiza viagens de ida e volta na rede.
• Opcionalmente, evita a verificação de restrições durante a carga.
• Utiliza o protocolo otimizado TDS bulk insert.
• Alcança um throughput comparável a bcp.exe e SqlBulkCopy.

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

Uso Básico

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

Importante

Se você criar ou alterar a tabela de destino na mesma sessão, chame conn.commit() antes de bulkcopy(). O protocolo de cópia em massa usa um canal interno separado para ler os metadados da tabela, por isso uma alteração DDL não confirmada pode causar um impasse ou 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	derivar	Tempo necessário para a operação em segundos.

Assinatura de 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 padrão, bulkcopy() mapeia colunas por posição ordinal. Cada coluna de dados corresponde à coluna da tabela no mesmo índice. Use o column_mappings parâmetro para sobrepor esse comportamento.

Lista de nomes das 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 pular ou reordenar as colunas:

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

Carregar a partir de arquivos

Você pode carregar dados de arquivos CSV e outros formatos de arquivo passando um gerador para bulkcopy().

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

Arquivos grandes com lotamento

Defina o batch_size parâmetro para controlar quantas linhas o driver envia por lote. Essa abordagem funciona bem para arquivos 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

Converta um DataFrame do pandas em uma lista de tuplas antes de passá-lo 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)

Lidar com valores NULL

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

Colunas de Identidade

Para inserir valores de 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 valor padrão), omita a coluna de identidade dos seus dados e use column_mappings para indicar as colunas que não são de identidade.

Opções de cópia em massa

Parâmetro	Default	Descrição
batch_size	0	Linhas por lote. 0 Permite que o servidor escolha o tamanho ideal.
timeout	30	Tempo limite da operação em segundos.
keep_identity	False	Preserve 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 em nível de tabela em vez de bloqueios em nível de linha.
keep_nulls	False	Preserve os valores NULL em vez de inserir os padrões das colunas.
fire_triggers	False	Dispare INSERT gatilhos na mesa de alvo.
use_internal_transaction	False	Envolva cada lote em uma transação interna.

Lidar com erros

bulkcopy() gera uma exceção se a carga falhar, então envolva a chamada em um try/except bloco para detectar erros. Lembre-se de que bulkcopy() é executado em sua própria conexão interna e confirma as linhas copiadas de forma independente, portanto um conn.rollback() na sua conexão principal não pode revertê-las. Para tornar um lote atômico, defina use_internal_transaction=True, que encapsula cada lote em sua própria transação, que é revertida 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 colocar o carregamento sob sua própria lógica de validação, copie em massa para uma tabela de preparação e, em seguida, mova as linhas para a tabela de destino com um INSERT ... SELECT dentro de uma transação em sua conexão principal. Isso INSERT é executado na sua conexão, portanto conn.rollback() o reverte se a validação falhar.

Authentication

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

Identidade gerenciada (ActiveDirectoryMSI)

Use Authentication=ActiveDirectoryMSI para identidade gerenciada atribuída ao sistema ou ao usuário. Esse método de autenticação é recomendado para serviços hospedados 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 gerenciada atribuída pelo usuário, passe o ID do cliente na cadeia de conexã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 padrão (ActiveDirectoryDefault)

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

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

Dicas de desempenho

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

Use geradores para grandes conjuntos de dados

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 travas de mesa para cargas mais rápidas

Quando não houver leitores simultâneos, configure table_lock=True para reduzir a sobrecarga de bloqueio durante grandes cargas iniciais.

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

Desabilite índices durante o carregamento

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

Carregue tabelas em paralelo

Abra uma conexão separada para cada tabela e execute as cargas simultaneamente.

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 a seguir compara cópia em massa com outros métodos de inserção de dados.

Método	Caso de uso	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() em um loop	Conjuntos de dados pequenos com lógica simples.	Mais lento

Conteúdo relacionado

• Execução de consultas
• Mapeamentos de tipo de dados
• Gerenciamento de transações
• Otimização do desempenho