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

Um DataFrame é colunar, por isso o caminho mais rápido é bulkcopy_arrow(), que consome a tabela de setas que os pandas já sabem produzir. bulkcopy()aceita tuplas de linha, por isso tens de achatar as colunas em objetos Python primeiro.

Converta a tabela Arrow para os tipos das colunas de destino antes de a carregar. pyarrow infere float64 para uma coluna numérica, que o condutor não pode mapear para dinheiro, decimal ou numérico:

import pandas as pd
import pyarrow as pa
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()

target = pa.schema([
    pa.field('ID', pa.int32()),
    pa.field('Name', pa.string()),
    pa.field('Amount', pa.decimal128(19, 4)),   # MONEY
])

table = pa.Table.from_pandas(df, preserve_index=False).cast(target)
result = cursor.bulkcopy_arrow("##PandasDemo", table)

Sem a conversão de tipo, o carregamento falha com ValueError: Cannot map Arrow column 'Amount' (Float64) to SQL column 'Amount' (Money). Constrói o cast com Table.cast() em vez de passar o esquema para Table.from_pandas(), que não pode converter uma coluna float para decimal128 diretamente. NaN os valores tornam-se SQL NULL neste caminho, por isso não precisas de os substituir primeiro.

Se precisares do caminho da linha-tupla, itertuples() já gera tuplas quando passas name=None:

data = list(df.itertuples(index=False, name=None))
result = cursor.bulkcopy("##PandasDemo", data)

Carregar dados do Apache Arrow

Uso cursor.bulkcopy_arrow() para carregar dados do Apache Arrow. Este método lê diretamente da memória Arrow, por isso não se constrói tuplas de linha em Python antes de a chamar.

O source argumento aceita a pyarrow.Table, a pyarrow.RecordBatch, a pyarrow.RecordBatchReader, ou qualquer objeto que exponha a interface de dados Arrow C. Os argumentos restantes são os mesmos que bulkcopy().

import mssql_python
import pyarrow as pa

conn = mssql_python.connect(connection_string)

# bulkcopy_arrow() opens its own connection, so commit the table creation first.
conn.autocommit = True
cursor = conn.cursor()

cursor.execute("""
    CREATE TABLE ##ArrowDemo (ID INT, Name NVARCHAR(50), Amount FLOAT)
""")

table = pa.table({
    "ID": pa.array([1, 2, 3], type=pa.int32()),
    "Name": pa.array(["Alice", "Bob", "Carol"], type=pa.string()),
    "Amount": pa.array([50000.0, 60000.0, 55000.0], type=pa.float64()),
})

result = cursor.bulkcopy_arrow("##ArrowDemo", table)
print(f"Copied {result['rows_copied']} rows")

Cada tipo de coluna Arrow deve ser compatível com o seu tipo de coluna SQL de destino. O autor não converte entre famílias de tipos, por isso passar uma float64 coluna para uma coluna de dinheiro aumenta ValueError antes de quaisquer linhas serem escritas. Use decimal128 para colunas de moeda, decimais e numéricas.

Passar uma fonte de Flecha para bulkcopy() eleva TypeError e direciona-te para bulkcopy_arrow().

Para mais informações sobre o suporte ao Arrow, incluindo como transmitir um conjunto de resultados de uma tabela para outra, veja integração com o Apache Arrow.

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. Aplica-se à operação de cópia em massa em si, não à ligação interna. Use 0 para desativar o tempo limite da operação.
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.

Note

bulkcopy() Abre uma ligação interna separada ao servidor. Essa ligação interna herda o tempo limite da consulta do cursor: defina Connection.timeout para um valor positivo antes de criar o cursor, e esse mesmo valor delimita a tentativa de ligação de cópia em massa. Se o tempo de espera da consulta do cursor for 0, a ligação interna usa o seu tempo limite padrão de ligação de 15 segundos. Um cursor assume o valor quando é criado, pelo que alterar Connection.timeout posteriormente não afeta um cursor existente nem uma operação de cópia em massa em curso. Aumente o tempo de espera da consulta antes de criar o cursor para endpoints lentos, limitados ou de alta latência (por exemplo, via VPN ou entre regiões).

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.

Comece a partir de uma fonte colunar

bulkcopy() recebe um iterável de tuplas de linhas, pelo que cada valor tem de existir sob a forma de um objeto Python antes de a cópia começar. Quando os dados já são colunares, bulkcopy_arrow() lê diretamente os buffers Arrow e salta essa etapa. Um DataFrame do pandas ou do Polars, um ficheiro Parquet e o resultado de cursor.arrow() são todos origens Arrow. Para mais informações, consulte Carregar dados do Apache Arrow.

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_arrow()	Grandes conjuntos de dados que já são colunares.	Mais rápido
cursor.bulkcopy()	Grandes conjuntos de dados (mais de 1.000 linhas) provenientes de fontes orientadas a linhas.	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

• Executar consultas com mssql-python
• Mapeamentos de tipos de dados para mssql-python
• Gestão de transações com mssql-python
• Ajuste de desempenho para aplicações mssql-python