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

Un DataFrame tiene estructura columnar, por lo que la vía más rápida es bulkcopy_arrow(), que consume la tabla Arrow que pandas ya sabe producir. bulkcopy()toma tuplas de fila, así que primero tienes que aplanar las columnas en objetos Python.

Convierte la tabla de flechas en los tipos de columnas de destino antes de cargarla. pyarrow infiere float64 para una columna numérica, que el conductor no puede asignar a dinero, decimal o 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)

Sin el cast, la carga falla con ValueError: Cannot map Arrow column 'Amount' (Float64) to SQL column 'Amount' (Money). Construye el cast con Table.cast() en lugar de pasar el esquema a Table.from_pandas(), que no puede convertir una columna flotante a decimal128 directamente. NaN los valores se convierten en SQL NULL en este camino, así que no necesitas reemplazarlos primero.

Si en su lugar necesitas la ruta de tuplas de fila, itertuples() ya devuelve tuplas cuando pasas name=None:

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

Cargar datos de Apache Arrow

Úsalo cursor.bulkcopy_arrow() para cargar los datos de Apache Arrow. Este método lee directamente de la memoria de Arrow, por lo que no se crean tuplas de filas de Python antes de llamar a este método.

El source argumento acepta un pyarrow.Table, un pyarrow.RecordBatch, un pyarrow.RecordBatchReader, o cualquier objeto que exponga la interfaz de datos Arrow C. Los argumentos restantes son los mismos 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 columna Arrow debe ser compatible con su tipo de columna SQL de destino. El escritor no convierte entre familias de tipos, así que pasar una columna float64 a una columna money genera ValueError antes de que se escriba ninguna fila. Usa decimal128 para columnas de dinero, decimales y valores numéricos.

Pasar una fuente de flechas a bulkcopy() eleva TypeError y te dirige a bulkcopy_arrow().

Para más información sobre el soporte de Arrow, incluyendo cómo transmitir un conjunto de resultados de una tabla a otra, véase integración con Apache Arrow.

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)

Utilice Authentication=ActiveDirectoryMSI para una identidad gestionada asignada por el sistema o 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.

Comienza desde una fuente columnar

bulkcopy() toma un iterable de tuplas de filas, por lo que cada valor debe estar presente como un objeto de Python antes de que comience la copia. Cuando los datos ya tienen formato columnar, bulkcopy_arrow() lee directamente los búferes de Arrow y omite ese paso. Un DataFrame de pandas o de Polars, un archivo Parquet y el resultado de cursor.arrow() son fuentes de Arrow. Para más información, véase Cargar datos de Apache Arrow.

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

• Ejecutar consultas con mssql-python
• Mapeos de tipos de datos para mssql-python
• Gestión de transacciones con mssql-python
• Ajuste del rendimiento para aplicaciones mssql-python