Carga de datos con combinación, inserción y anexión

Después de transformar y limpiar los datos, el último paso en la canalización es cargarlos en las tablas de destino. Azure Databricks proporciona tres estrategias de carga principales: anexar, sobrescribir y combinar. Cada estrategia sirve diferentes casos de uso y elegir el adecuado depende de cómo se relacionan los datos de origen con los registros existentes.

En esta unidad, aprenderá a anexar nuevas filas con INSERT, reemplazar los datos por operaciones de sobrescritura y sincronizar los cambios mediante MERGE para escenarios upsert.

Anexar datos con INSERT INTO

La operación de carga más sencilla agrega nuevas filas a una tabla existente sin modificar los datos actuales. Use INSERT INTO cuando cargue lotes de datos nuevos que no se superpongan con registros existentes, como registros de transacciones diarios, flujos de eventos nuevos o extractos de datos incrementales.

INSERT INTO funciona con SQL y PySpark. En los ejemplos siguientes se muestra cómo agregar filas a una tabla de ventas.

-- Append new records using VALUES
INSERT INTO sales.transactions (transaction_id, amount, transaction_date)
VALUES 
    ('TXN001', 150.00, '2026-01-15'),
    ('TXN002', 275.50, '2026-01-15');

-- Append from a staging table or query
INSERT INTO sales.transactions
SELECT * FROM staging.daily_transactions
WHERE transaction_date = current_date();

En PySpark, los datos se agregan utilizando un DataFrame en el modo append.

# Append DataFrame to existing table
df_new_transactions.write.mode("append").saveAsTable("sales.transactions")

# Or use insertInto for existing tables
df_new_transactions.write.insertInto("sales.transactions")

Sustituir datos con INSERT OVERWRITE

A veces, debe reemplazar los datos existentes en lugar de agregarlos. INSERT OVERWRITE trunca la tabla o partición de destino antes de insertar nuevas filas. Este enfoque funciona bien para volver a procesar lotes con errores, actualizar tablas de búsqueda o volver a generar agregaciones.

-- Replace entire table contents
INSERT OVERWRITE sales.daily_summary
SELECT 
    region,
    SUM(amount) as total_sales,
    COUNT(*) as transaction_count
FROM sales.transactions
WHERE transaction_date = current_date()
GROUP BY region;

En el caso de las tablas con particiones, puede sobrescribir particiones específicas mientras deja intactas otras:

-- Replace only January 2026 partition
INSERT OVERWRITE sales.monthly_report
PARTITION (report_month = '2026-01')
SELECT region, product_category, SUM(amount) as revenue
FROM sales.transactions
WHERE transaction_date BETWEEN '2026-01-01' AND '2026-01-31'
GROUP BY region, product_category;

Delta Lake también admite sobrescrituras selectivas con la REPLACE WHERE cláusula , que elimina las filas que coinciden con una condición antes de insertar:

-- Replace transactions for a specific date range
INSERT INTO sales.transactions
REPLACE WHERE transaction_date >= '2026-01-01' AND transaction_date <= '2026-01-07'
SELECT * FROM staging.corrected_transactions;

De forma predeterminada, cada fila que devuelve la consulta debe coincidir con el predicado REPLACE WHERE. Si alguna fila está fuera del predicado, la instrucción falla y no escribe nada.

En PySpark, use overwrite el modo para reemplazar la tabla completa y agregue la replaceWhere opción para reemplazar solo las filas que coinciden con una condición:

# Overwrite entire table
df_refreshed.write.mode("overwrite").saveAsTable("sales.daily_summary")

# Replace only the January 2026 rows
(df_january.write
    .mode("overwrite")
    .option("replaceWhere", "report_month = '2026-01'")
    .saveAsTable("sales.monthly_report"))

Reemplazar dinámicamente los datos con REPLACE USING

REPLACE USING es el enfoque recomendado para sobrescribir datos dinámicos. Funciona en todos los tipos de proceso, incluidos los almacenes de SQL de Databricks y el proceso sin servidor, y no requiere la configuración de sesión de Spark. A diferencia partitionOverwriteModede , REPLACE USING admite tablas con particiones, tablas sin particiones y tablas con clústeres líquidos.

-- Replace only the partitions touched by the incoming data
INSERT INTO sales.monthly_report
  REPLACE USING (report_month)
  SELECT region, product_category, SUM(amount) AS revenue, report_month
  FROM staging.corrected_transactions
  GROUP BY region, product_category, report_month;

\ elimina de forma atómica las filas en \ que coinciden con las filas entrantes en \, y a continuación, inserta las nuevas filas. Las filas que no coinciden con las columnas especificadas se dejan intactas. En Databricks Runtime 16.3–17.1, las USING columnas deben ser el conjunto completo de columnas de partición de la tabla (comportamiento heredado). En Databricks Runtime 17.2 y versiones posteriores, se admiten todas las columnas de cualquier tipo de tabla, incluidas las tablas sin particionar y las tablas con clústeres líquidos. La cláusula SQL REPLACE USING requiere Databricks Runtime 16.3 o posterior, y la opción Python y Scala replaceUsing requiere Databricks Runtime 18.2 o superior. Dado que REPLACE USING usa la igualdad normal, las filas con valores de NULL en las columnas USING no coinciden y no se quitan de la tabla de destino.

Cuando necesite coincidencias más flexibles, como reemplazar filas en función de una condición personalizada o tratar valores NULL como iguales, use REPLACE ON en su lugar:

-- Replace rows matching a user-defined condition across source and target
INSERT INTO sales.transactions AS target
  REPLACE ON (target.transaction_id = source.transaction_id)
  (SELECT * FROM staging.corrected_transactions) AS source;

REPLACE ONrequiere Databricks Runtime 17.1 y versiones posteriores para SQL y Databricks Runtime 18.2 y versiones posteriores para la opción Python y ScalareplaceOn. Utilice REPLACE WHERE (cubierto anteriormente) cuando necesite un reemplazo según condiciones sin referencia a una tabla de origen.

Combinar datos para operaciones upsert

Las canalizaciones de datos reales suelen necesitar actualizar los registros existentes al insertar nuevos. La instrucción MERGE controla este patrón(normalmente denominado "upsert"), haciendo coincidir las filas de origen con las filas de destino y aplicando diferentes acciones en función de si existe una coincidencia.

MERGE es esencial para las dimensiones que cambian lentamente, sincronizar datos de sistemas de origen y procesar fuentes de captura de datos modificados (CDC).

MERGE INTO customers AS target
USING customer_updates AS source
ON target.customer_id = source.customer_id
WHEN MATCHED THEN
    UPDATE SET 
        target.email = source.email,
        target.phone = source.phone,
        target.last_updated = current_timestamp()
WHEN NOT MATCHED THEN
    INSERT (customer_id, name, email, phone, created_date, last_updated)
    VALUES (source.customer_id, source.name, source.email, source.phone, 
            current_timestamp(), current_timestamp());

Esta consulta actualiza el correo electrónico y el teléfono de los clientes existentes al insertar registros de clientes completamente nuevos. La cláusula ON define la clave coincidente, normalmente una clave principal o un identificador de negocio.

Combinación con lógica condicional

Agregue condiciones a las cláusulas WHEN para controlar escenarios más complejos. Por ejemplo, es posible que solo actualice los registros que realmente han cambiado:

MERGE INTO products AS target
USING product_feed AS source
ON target.sku = source.sku
WHEN MATCHED AND source.price <> target.price THEN
    UPDATE SET target.price = source.price, target.updated_at = current_timestamp()
WHEN MATCHED AND source.discontinued = true THEN
    DELETE
WHEN NOT MATCHED THEN
    INSERT *;

La INSERT * sintaxis inserta todas las columnas del origen cuando coinciden con el esquema de destino.

Combinar con PySpark

Para el control mediante programación, use la API de Python de Delta Lake:

from delta.tables import DeltaTable

target_table = DeltaTable.forName(spark, "customers")

target_table.alias("target").merge(
    source_df.alias("source"),
    "target.customer_id = source.customer_id"
).whenMatchedUpdate(set={
    "email": "source.email",
    "phone": "source.phone",
    "last_updated": "current_timestamp()"
}).whenNotMatchedInsert(values={
    "customer_id": "source.customer_id",
    "name": "source.name",
    "email": "source.email",
    "phone": "source.phone",
    "created_date": "current_timestamp()",
    "last_updated": "current_timestamp()"
}).execute()

Para casos más sencillos en los que coinciden los esquemas de origen y destino, use los métodos de conveniencia:

target_table.alias("target").merge(
    source_df.alias("source"),
    "target.customer_id = source.customer_id"
).whenMatchedUpdateAll(
).whenNotMatchedInsertAll(
).execute()

Elección de la estrategia de carga correcta

La elección de la operación de carga depende de la relación entre los datos de origen y de destino:

Comprender estos patrones de carga le prepara para crear canalizaciones de datos confiables que mantienen la integridad de los datos, a la vez que procesan de forma eficaz registros nuevos y modificados.