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 as viagens de ida e volta da 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:

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 retornado

bulkcopy() retorna um dicionário:

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 envio em lote

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

Carregar DataFrames do pandas

Um DataFrame é colunar, então o caminho mais rápido é bulkcopy_arrow(), que consome a tabela Arrow que o pandas já sabe produzir. bulkcopy() recebe tuplas de linhas; então, primeiro você precisa nivelar as colunas em objetos Python.

Converta a tabela Arrow para os tipos das colunas de destino antes de carregá-la. pyarrow infere float64 para uma coluna numérica, que o motorista 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). Construa o cast com Table.cast() em vez de passar o esquema para Table.from_pandas(), que não pode converter uma coluna flutuante para decimal128 diretamente. NaN valores se tornam SQL NULL nesse caminho, então você não precisa substituí-los primeiro.

Se você precisar do caminho da linha-tupla, itertuples() já gera tuplas quando você passa name=None:

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

Carregar dados do Apache Arrow

Use cursor.bulkcopy_arrow() para carregar dados do Apache Arrow. Esse método lê diretamente da memória Arrow, então você não precisa construir tuplas de linhas em Python antes de chamá-lo.

O source argumento aceita um pyarrow.Table, a pyarrow.RecordBatch, um pyarrow.RecordBatchReader, ou qualquer objeto que exponha a interface de dados Arrow C. Os demais argumentos 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 tipo de coluna SQL de destino correspondente. O escritor não converte entre famílias de tipos; então, passar uma coluna float64 para uma coluna money gera ValueError antes da gravação de linhas. Use decimal128 para colunas de dinheiro, decimais e numéricas.

Passar uma fonte de Flecha para bulkcopy() eleva TypeError e direciona você 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.

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

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 entidade 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.

Comece a partir de uma fonte colunar

bulkcopy() recebe um iterável de tuplas de linhas, portanto cada valor precisa existir como um objeto Python antes do início da cópia. Quando os dados já são colunares, bulkcopy_arrow() ele lê os buffers do Arrow diretamente e pula essa etapa. Um DataFrame do pandas ou do Polars, um arquivo Parquet e o resultado de cursor.arrow() são todos fontes do Arrow. Para mais informações, veja Carregar dados do Apache Arrow.

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.

Conteúdo relacionado

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