Flujos de REPLACE WHERE para tablas de streaming independientes

Los flujos REPLACE WHERE vuelven a calcular y sobrescriben un subconjunto específico de una tabla de streaming independiente sin volver a procesar todo el historial de la tabla. Controlan los datos de llegada tardía, el reprocesamiento ascendente, la evolución del esquema y los rellenamientos pendientes.

Con un flujo REPLACE WHERE , se define un predicado en la tabla de destino. Todas las filas que coinciden con el predicado se eliminan y se sustituyen volviendo a evaluar la consulta de origen para ese mismo rango del predicado. Las filas que no coinciden con el predicado se dejan intactas.

Requirements

Los flujos REPLACE WHERE tienen los siguientes requisitos:

  • Databricks recomienda Unity Catalog y la computación sin servidor. La actualización incremental solo se admite en el proceso de cálculo sin servidor.

Cuándo usar flujos REPLACE WHERE

Use flujos REPLACE WHERE para los escenarios siguientes:

  • Procesamiento por lotes incremental sin semántica de streaming: procese nuevas filas en lotes sin administrar conceptos de streaming como marcas de agua.
  • Reprocesamiento selectivo: recompute solo las filas que coinciden con un predicado mientras dejan sin modificar todas las demás filas.
  • Escenarios más allá de las funcionalidades de vista materializadas estándar:
    • Tablas de destino con retención más larga que el origen
    • Impedir la recomputación cuando cambia una tabla de dimensiones
    • Evolución del esquema sin volver a calcular todo el historial

Crear un flujo REPLACE WHERE

Use la cláusula FLOW REPLACE WHERE en línea con CREATE OR REFRESH STREAMING TABLE:

CREATE OR REFRESH STREAMING TABLE orders_enriched
SCHEDULE EVERY 1 DAY
FLOW REPLACE WHERE date >= date_add(current_date(), -7) BY NAME
SELECT
  o.order_id,
  o.date,
  o.region,
  p.product_name,
  o.qty,
  o.price
FROM orders_fct o
JOIN product_dim p
  ON o.product_id = p.product_id;

Durante la actualización, se eliminan todas las filas de la tabla de destino que coinciden con el predicado, se vuelve a calcular la consulta de origen para ese mismo intervalo de predicado y se insertan los nuevos resultados. En este ejemplo, todas las filas de los últimos 7 días se eliminan de orders_enriched y se recalculan usando la consulta de origen.

No es necesario agregar el predicado a la consulta de origen. El motor de canalización lo aplica automáticamente al leer desde el origen.

Note

BY NAME es obligatorio. Garantiza que las columnas coincidan por nombre en lugar de por posición.

Reposición de datos históricos

Para escribir filas históricas o corregidas en la tabla de destino fuera de las actualizaciones programadas, elija entre dos mecanismos en función de dónde residen los datos históricos:

  • Anulaciones de predicado: vuelva a ejecutar la consulta de origen del flujo para un rango de predicado de una sola vez usando REFRESH STREAMING TABLE ... WHERE. Use cuando los datos históricos proceden del mismo origen que los datos incrementales.
  • Instrucciones DML: Insertar directamente en la tabla de destino, omitiendo el flujo. Use cuando los datos históricos residen en un origen diferente al de los datos incrementales.

Anulaciones de predicados

Invalide el predicado del REPLACE WHERE flujo para una sola actualización sin modificar la definición de la tabla. Los reemplazos de predicado son de un solo uso, solo se aplican a esa actualización y no afectan a las actualizaciones programadas posteriores.

Utilice la cláusula WHERE con REFRESH STREAMING TABLE:

REFRESH STREAMING TABLE orders_enriched WHERE date BETWEEN '2020-01-01' AND '2024-12-31';

El flujo elimina solo las filas que coinciden con el predicado de anulación y las vuelve a calcular a partir del origen. El predicado estático del flujo no cambia, por lo que la siguiente actualización programada usa el predicado original.

Puede combinar la anulación con SYNC o ASYNC. Por ejemplo, para iniciar la actualización en segundo plano y devolver de inmediato un enlace a la actualización del pipeline:

REFRESH STREAMING TABLE orders_enriched WHERE date >= date_add(current_date(), -30) ASYNC;

Note

La cláusula WHERE solo se admite en una tabla de flujo creada con una cláusula FLOW REPLACE WHERE. Si se especifica en cualquier otra tabla de streaming, se devuelve un error. Consulta REFRESH (MATERIALIZED VIEW o STREAMING TABLE).

Instrucciones DML

Ejecute instrucciones DML directamente en la tabla de destino para cargar filas que proceden de un origen diferente al flujo:

INSERT INTO orders_enriched
SELECT *
FROM orders_enriched_legacy
WHERE date < '2025-01-01';

Comportamiento de actualización completa

Una actualización completa de un flujo REPLACE WHERE vuelve a ejecutar la consulta de origen con solo el predicado definido del flujo. Elimina permanentemente todas las filas de la tabla de destino que no cumplan ese predicado, incluidas las filas que las anulaciones del predicado o las instrucciones DML hayan insertado anteriormente fuera del rango de predicado definido.

Advertencia

Una actualización completa borra todos los datos existentes y vuelve a ejecutar el flujo con solo su predicado definido. Si una canalización ha estado ejecutándose durante un año con un predicado de 7 días, una actualización completa hace que la tabla contenga solo los datos de los últimos 7 días. Todas las filas anteriores se eliminan permanentemente.

REFRESH STREAMING TABLE orders_enriched FULL;

Para evitar actualizaciones completas en una tabla, establezca la propiedad de tabla pipelines.reset.allowed en false:

CREATE OR REFRESH STREAMING TABLE orders_enriched
  TBLPROPERTIES (pipelines.reset.allowed = 'false')
  FLOW REPLACE WHERE date >= date_add(current_date(), -7) BY NAME
  ...

Actualización incremental

Los flujos REPLACE WHERE usan la actualización incremental siempre que sea posible, reprocesando solo los datos de origen que han cambiado desde la última actualización en lugar de volver a calcular toda la ventana de reemplazo. La actualización incremental requiere un proceso sin servidor.

Cuando se aplica la actualización incremental

Todo lo siguiente debe ser verdadero:

  • La canalización se ejecuta en proceso sin servidor.
  • Se admite la forma de consulta. Consulte Actualización incremental para el conjunto de operadores admitidos.
  • El predicado hace referencia a columnas base de una tabla de origen. Los predicados sobre valores derivados, como los resultados de funciones de agregado o de ventana, no se pueden aplicar en el origen, lo que desactiva la actualización incremental.
  • Ninguna fila ha sido modificada por DML externo en la ventana de reemplazo actual. DML que modifica filas fuera de la ventana actual no se ve afectada.
  • La ventana de reemplazo actual no incluye filas excluidas del predicado anterior. Si se amplía el predicado para abarcar un intervalo que no se había procesado previamente, esa actualización recurre de nuevo a la recomputación completa. Las actualizaciones posteriores son aptas para la actualización incremental de nuevo.
  • El predicado es determinista. Los predicados que usan funciones no deterministas, como rand(), deshabilitan la actualización incremental. Se permiten funciones temporales como current_date() .

La primera actualización de cualquier flujo siempre es un cálculo completo. Si no se cumple alguna de las condiciones, dicha actualización recurre a un recálculo completo de la ventana de sustitución actual.

Procedimientos recomendados para la actualización incremental

Siga estas instrucciones para que los flujos REPLACE WHERE sigan siendo aptos para la actualización incremental.

Usar un límite inferior móvil

Los predicados con un límite inferior móvil siguen siendo aptos para la actualización incremental indefinidamente.

FLOW REPLACE WHERE date >= date_add(current_date(), -7)

Un límite superior móvil, como date BETWEEN date_add(current_date(), -7) AND current_date(), puede desplazar la ventana para incluir filas que antes quedaban excluidas, lo que provoca un recurso puntual al recálculo completo.

Incluir la columna de predicado en GROUP BY

Al realizar una agregación, incluya la columna del predicado en GROUP BY para que el motor pueda empujar el predicado por debajo de la agregación.

FLOW REPLACE WHERE date >= date_add(current_date(), -7) BY NAME
SELECT date, region, SUM(amount) AS total
FROM sales
GROUP BY date, region;

Si falta la columna del predicado en GROUP BY, el predicado no se puede desplazar por debajo de la agregación y la fuente se escanea por completo.

Incluir la columna de predicado en claves de combinación

Incluya la columna del predicado en la condición de unión para que el motor pueda descartar todas las fuentes unidas.

FLOW REPLACE WHERE date >= date_add(current_date(), -7) BY NAME
SELECT f.date, f.user_id, d.region, f.revenue
FROM fact f
JOIN dim d ON f.date = d.date AND f.user_id = d.user_id;

Si una tabla combinada no expone la columna de predicado, esa tabla se examina en su totalidad en cada actualización.

Diagnosticar el recurso a la recomputación completa

Cuando una actualización recurre a un recálculo completo, el motivo se informa en el evento planning_information del flujo. Consulte Supervisión de los registros de eventos de canalización. En la tabla siguiente se enumeran los motivos notificados en el evento:

Motivo Meaning
EXTERNAL_CHANGE_IN_REPLACE_WINDOW Una instrucción DML externa modificó filas en la ventana de reemplazo actual.
REPLACE_WHERE_NOT_DETERMINISTIC El predicado usa expresiones no deterministas.
PRIOR_REPLACE_WHERE_NOT_DETERMINISTIC La actualización anterior usó un predicado no determinista.
UNSUPPORTED_REPLACE_WHERE_PREDICATE El predicado no se puede aplicar a ningún origen, la ventana actual contiene filas no procesadas por el predicado anterior o la ejecución usa una anulación del predicado.

Ejemplos

En los ejemplos siguientes se muestran patrones de flujo REPLACE WHERE comunes.

Ejemplo 1: Mantener agregados históricos de un origen de retención limitado

Este ejemplo mantiene agregados diarios de forma indefinida, incluso después de que los datos brutos caduquen en la tabla de origen (retención de 3 días):

CREATE OR REFRESH STREAMING TABLE events_agg
FLOW REPLACE WHERE date >= date_add(current_date(), -3) BY NAME
SELECT
  date,
  key,
  SUM(val) AS agg
FROM events_raw
GROUP BY ALL;

Ejemplo 2: Evitar la recomputación cuando cambia una tabla de dimensiones

En este ejemplo se mantienen las filas de hechos históricas sin cambios cuando cambian los atributos de dimensión:

CREATE OR REFRESH STREAMING TABLE fact_dim_join
FLOW REPLACE WHERE f.date >= date_add(current_date(), -1) BY NAME
SELECT
  f.date,
  f.user_id,
  d.region,
  f.revenue
FROM fact_table f
JOIN dim_users d
  ON f.user_id = d.user_id;

Si cambia la región de un usuario, solo se vuelven a calcular las filas recientes. Las filas históricas conservan el valor de región en el momento en que se escribieron.

Ejemplo 3: Agregar una nueva métrica sin volver a calcular el historial completo

En este ejemplo se muestra cómo evolucionar una definición de tabla y rellenar solo un intervalo de destino:

  1. Defina la tabla inicial:

    CREATE OR REFRESH STREAMING TABLE clickstream_daily
FLOW REPLACE WHERE event_date >= date_add(current_date(), -7) BY NAME
SELECT
  event_date,
  page_id,
  COUNT(*) AS clicks
FROM clickstream_raw
GROUP BY ALL;

  2. Actualice la consulta para agregar uniq_users:

    CREATE OR REFRESH STREAMING TABLE clickstream_daily
FLOW REPLACE WHERE event_date >= date_add(current_date(), -7) BY NAME
SELECT
  event_date,
  page_id,
  COUNT(*) AS clicks,
  COUNT(DISTINCT user_id) AS uniq_users
FROM clickstream_raw
GROUP BY ALL;

    Las filas anteriores a la ventana de 7 días contienen NULL para uniq_users.

Ejemplo 4: Iteración en una ventana pequeña antes de rellenar el historial completo

En este ejemplo se muestra cómo validar la lógica de consulta en una ventana de datos pequeña antes de procesar el intervalo histórico completo.

Comience con una ventana corta para validar las métricas e iterar en la lógica de negocios con menores costos de proceso:

CREATE OR REFRESH STREAMING TABLE revenue_attribution
FLOW REPLACE WHERE event_date >= date_add(current_date(), -7) BY NAME
SELECT
  event_date,
  campaign_id,
  SUM(revenue) AS total_revenue
FROM marketing_events
GROUP BY ALL;

Una ventana corta recalcula solo los últimos 7 días en cada actualización, por lo que revise la consulta tantas veces como sea necesario antes de lanzar una ejecución histórica completa.

Una vez finalizada la consulta, use DML para rellenar el intervalo histórico completo:

INSERT INTO revenue_attribution
SELECT
  event_date,
  campaign_id,
  SUM(revenue) AS total_revenue
FROM marketing_events
WHERE event_date < date_add(current_date(), -7)
GROUP BY ALL;
  • Last updated on