Usa copia masiva con mssql-python

El controlador mssql-python incluye una función de copia masiva que inserta de forma eficiente grandes cantidades de datos en SQL Server, Azure SQL Database, Azure SQL Managed Instance y base de datos SQL en Microsoft Fabric.

El cursor.bulkcopy() método proporciona una ruta de alto rendimiento para cargar grandes conjuntos de datos:

• Minimiza los viajes de ida y vuelta por la red.
• Opcionalmente, se evita la comprobación de restricciones durante la carga.
• Utiliza el protocolo optimizado TDS bulk insert.
• Logra un rendimiento comparable a bcp.exe y SqlBulkCopy.

La extensión nativa basada mssql_py_core en Rust impulsa la función de copia masiva. Se ejecuta fuera del flujo normal del cursor execute().

Uso básico

Llama a bulkcopy() en un cursor, pasando el nombre de la tabla de destino y un iterable de tuplas de fila u 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 devuelto

bulkcopy() devuelve un diccionario:

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

Asignaciones de columnas

Por defecto, bulkcopy() mapea las columnas por posición ordinal. Cada columna de datos se corresponde con la columna de la tabla con el mismo índice. Usa el column_mappings parámetro para anular este comportamiento.

Lista de nombres de columnas

Cada posición en la lista corresponde al índice de datos fuente:

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

Formato avanzado: mapeo explícito de índices

Cada tupla adopta la forma (source_index, target_column_name). Utiliza este formato para saltar o reordenar columnas:

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

Cargar desde archivos

Puedes cargar datos desde archivos CSV y otros formatos de archivo pasando un generador a bulkcopy().

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

Archivos grandes con procesamiento por lotes

Configura el batch_size parámetro para controlar cuántas filas envía el controlador por lote. Este enfoque funciona bien para archivos 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")

Cargar DataFrames de Pandas

Convierte un DataFrame de pandas en una lista de tuplas antes de pasarla a 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)

Manejar valores NULL

Pase None en cualquier posición de columna para insertar un 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)

Columnas de identidad

Para insertar valores identidad explícitos, establezca 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)

Cuando se usa keep_identity=False (opción predeterminada), omita la columna de identidad de los datos y use column_mappings para dirigirse a las columnas que no son de identidad.

Opciones de copia masiva

Gestión de errores

bulkcopy() abre una excepción si la carga falla, así que envuelve la llamada en un try/except bloque para detectar errores. Ten en cuenta que bulkcopy() se ejecuta en su propia conexión interna y confirma las filas copiadas de forma independiente, así que un conn.rollback() en tu conexión principal no puede deshacer esos cambios. Para hacer que un lote sea atómico, establezca use_internal_transaction=True, lo que hace que cada lote quede envuelto en su propia transacción, que se revierte automáticamente si el lote produce un error:

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 hacer que una carga pase por tu propia lógica de validación, haz una copia masiva en una tabla intermedia y luego mueve las filas a la tabla de destino con un INSERT ... SELECT dentro de una transacción en tu conexión principal. Eso INSERT se ejecuta en tu conexión, así que conn.rollback() lo deshace si falla la validación.

Autenticación

La copia masiva utiliza un canal interno separado que requiere su propio token. El controlador gestiona automáticamente la adquisición de tokens para los métodos de autenticación compatibles.

Identidad gestionada (ActiveDirectoryMSI)

Use Authentication=ActiveDirectoryMSI para la identidad administrada asignada por el sistema o asignada por el usuario. Este método de autenticación se recomienda para servicios alojados en Azure como máquinas virtuales de Azure, App Service, Functions y 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 una identidad gestionada asignada por el usuario, pasa el ID del cliente en la string de conexión:

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

Entidad de servicio (ActiveDirectoryServicePrincipal)

Use Authentication=ActiveDirectoryServicePrincipal para la autenticación de entidad de servicio (credenciales de 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")

Cadena de credenciales predeterminada (ActiveDirectoryDefault)

ActiveDirectoryDefault Prueba varios proveedores de credenciales en secuencia, como variables de entorno, identidad de carga de trabajo, identidad gestionada y más. Funciona tanto para el desarrollo local como para servicios alojados en Azure sin cambios de código.

Para más información sobre autenticación, consulte Microsoft Entra authentication.

Consejos de rendimiento

Las siguientes técnicas te ayudan a maximizar el rendimiento de copias masivas.

Uso de generadores para grandes conjuntos de datos

Los generadores minimizan el uso de memoria porque bulkcopy() aceptan cualquier iterable:

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

Usa cerraduras de mesa para cargas más rápidas

Cuando no tengas lectores concurrentes, configura table_lock=True para reducir la sobrecarga de bloqueo durante cargas iniciales grandes.

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

Desactivar los índices durante la carga

Desactiva temporalmente los índices no agrupados antes de la carga masiva y reconstruyelos después para mejorar el rendimiento:

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

Cargar tablas en paralelo

Abre una conexión separada para cada tabla y ejecuta las cargas simultáneamente.

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

Comparación con alternativas

La siguiente tabla compara la copia masiva con otros métodos de inserción de datos.

Contenido relacionado

• Ejecución de consultas
• Asignaciones de tipos de datos
• Administración de transacciones
• Ajuste del rendimiento