Flux REPLACE WHERE pour les tables de streaming autonomes

Les flux REPLACE WHERE recalculent et écrasent un sous-ensemble ciblé d'une table de streaming autonome sans retraiter tout l'historique de la table. Ils traitent les données arrivées en retard, le retraitement en amont, l'évolution des schémas et le remplissage.

Avec un flux REPLACE WHERE, vous définissez un prédicat sur la table cible. Toutes les lignes correspondant au prédicat sont supprimées et remplacées par une réévaluation de la query source pour cette même plage de prédicats. Les lignes qui ne correspondent pas au prédicat restent intactes.

Configuration requise

Les flux REPLACE WHERE doivent répondre aux exigences suivantes :

Databricks recommande Unity Catalog et le compute serverless. Le refresh incrémentiel n’est prise en charge que par le compute Serverless.

Quand utiliser les flux REPLACE WHERE

Utilisez des flux REPLACE WHERE dans les scénarios suivants :

Traitement par batch incrémentiel sans sémantique de streaming : Traitez les nouvelles lignes par batch sans gérer de concepts de streaming tels que les filigranes.
Reprotraitement sélectif : recalculez uniquement les lignes qui correspondent à un prédicat tout en laissant toutes les autres lignes intactes.
Scénarios allant au-delà des capacités des vues matérialisées standard :
- Tables de destination avec une rétention plus longue que la source
- Empêcher le recalcul lorsqu’une table de dimension change
- Évolution des schémas sans recalculer tout l'historique

Créez un flux REPLACE WHERE

Utilisez la clause FLOW REPLACE WHERE en ligne avec CREATE OR REFRESH STREAMING TABLE:

SQL
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;

Lors du refresh, toutes les lignes de la table cible correspondant au prédicat sont supprimées, la query source est recalculée pour cette même plage de prédicats et les nouveaux résultats sont insérés. Dans cet exemple, toutes les lignes des 7 derniers jours sont supprimées de orders_enriched et recalculées à l’aide de la query source.

Sinon, utilisez la syntaxe longue CREATE FLOW pour définir la table de streaming et le flux REPLACE WHERE dans des instructions distinctes :

SQL
CREATE STREAMING TABLE orders_enriched;

CREATE FLOW orders_enriched AS
INSERT INTO orders_enriched BY NAME
REPLACE WHERE date >= date_add(current_date(), -7)
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;

Les deux formulaires sont équivalents. Pour la référence complète de CREATE FLOW, voir CREATE FLOW (pipelines).

Vous n'avez pas besoin d'ajouter le prédicat à la query source. Le moteur de pipeline l’applique automatiquement lors de la lecture à partir de la source.

remarque

BY NAME est requis. Cela garantit que les colonnes sont associées par leur nom plutôt que par leur position.

Remplissage rétroactif des données historiques

Pour écrire des lignes historiques ou corrigées dans la table cible en dehors des refresh planifiés, choisissez entre deux mécanismes selon l'emplacement des données historiques :

Remplacements de prédicats : relancez la query source du flux pour une plage de prédicats ponctuelle en utilisant REFRESH STREAMING TABLE ... WHERE. À utiliser lorsque les données historiques proviennent de la même source que les données incrémentielles.
Instructions DML : Insérez directement dans la table cible en contournant le flux. À utiliser lorsque les données historiques résident dans une source différente des données incrémentielles.

Remplacements de prédicats

Remplacez le prédicat REPLACE WHERE du flux pour un seul refresh sans modifier la définition de table. Les remplacements de prédicat sont uniques, s’appliquent uniquement à cette refresh et n’affectent pas les refresh planifiées suivantes.

Utilisez la clause WHERE avec REFRESH STREAMING TABLE:

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

Le flux supprime uniquement les lignes correspondant au prédicat de remplacement et les recalcule à partir de la source. Le prédicat statique du flux n’est pas modifié, de sorte que le prochain refresh planifié utilise le prédicat d’origine.

Vous pouvez combiner le remplacement avec SYNC ou ASYNC. Par exemple, pour start l'refresh en arrière-plan et revenir immédiatement avec un lien vers la mise à jour du pipeline :

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

remarque

La clause WHERE n'est prise en charge que sur une table de streaming créée avec une clause FLOW REPLACE WHERE. La spécification sur toute autre table de streaming renvoie une erreur. Voir REFRESH (MATERIALIZED VIEW or STREAMING TABLE).

Instructions DML

Exécutez des instructions DML directement sur la table cible pour charger des lignes provenant d'une source différente de celle du flux :

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

Comportement de refresh complète

Une refresh complète d'un flux REPLACE WHERE réexécute la query source en utilisant uniquement le prédicat défini du flux. Cela supprime définitivement chaque ligne de la cible qui ne correspond pas à ce prédicat, y compris les lignes que ce prédicat remplace ou les instructions DML précédemment insérées en dehors de la plage de prédicats définie.

attention

Une refresh complète efface toutes les données existantes et exécute à nouveau le flux en utilisant uniquement son prédicat défini. Si un pipeline s’exécute depuis un an avec un prédicat de 7 jours, une refresh complète entraîne la table à ne contenir que les 7 derniers jours de données. Toutes les anciennes lignes sont définitivement supprimées.

SQL
REFRESH STREAMING TABLE orders_enriched FULL;

Pour éviter les actualisations complètes sur une table, définissez la propriété de table pipelines.reset.allowed sur false:

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

Actualisation refresh

Les flux REPLACE WHERE utilisent une refresh incrémentielle lorsque cela est possible, en ne retraitant que les données source qui ont changé depuis la dernière refresh au lieu de recalculer l'intégralité de la fenêtre de remplacement. Le refresh incrémentiel nécessite un compute Serverless.

Quand l'refresh incrémentiel s'applique

Tout ce qui suit doit être vrai :

Le pipeline s'exécute sur un compute serverless.
La forme de la query est prise en charge. Voir Actualisation refresh pour l'ensemble d'opérateurs pris en charge.
Le prédicat fait référence aux colonnes de base d’une table source. Les prédicats sur les valeurs dérivées, telles que les sorties de fonctions agrégées ou de fenêtres, ne peuvent pas être poussés vers une source, ce qui désactive le refresh incrémentiel.
Aucune instruction DML externe n’a modifié des lignes dans la fenêtre de remplacement actuelle. Le DML qui modifie des lignes en dehors de la fenêtre actuelle n’est pas affecté.
La fenêtre de remplacement actuelle n’inclut pas les lignes que le prédicat précédent a exclues. Si vous élargissez le prédicat pour couvrir une plage non traitée précédemment, cette refresh revient à un recalcul complet. Les refresh suivantes sont à nouveau éligibles au refresh incrémentiel.
Le prédicat est déterministe. Les prédicats utilisant des fonctions non déterministes telles que rand() désactivent le refresh incrémentiel. Les fonctions temporelles telles que current_date() sont autorisées.

La première refresh d’un flux est toujours un calcul complet. Si une condition n’est pas remplie, cette refresh revient au calcul complet de la fenêtre de remplacement actuelle.

Meilleures pratiques pour l’refresh incrémentiel

Suivez ces directives pour que les flux REPLACE WHERE restent éligibles à l’actualisation incrémentielle refresh.

Utiliser une limite inférieure mobile

Les prédicats avec une borne inférieure mobile restent éligibles pour un refresh incrémentiel indéfiniment.

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

Une limite supérieure mobile, telle que date BETWEEN date_add(current_date(), -7) AND current_date(), peut décaler la fenêtre pour inclure des lignes précédemment exclues, déclenchant un fallback ponctuel vers un recalcul complet.

Inclure la colonne prédicat dans GROUP BY

Lors de l’agrégation, incluez la colonne de prédicat dans GROUP BY afin que le moteur puisse pousser le prédicat sous l’agrégation.

SQL
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 la colonne prédicat est absente de GROUP BY, le prédicat ne peut pas être poussé en dessous de l'agrégation et la source est analysée dans sa totalité.

Inclure la colonne de prédicat dans les clés de jointure

Incluez la colonne de prédicat dans la condition de jointure afin que le moteur puisse élaguer toutes les sources jointes.

SQL
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 une table jointe n'expose pas la colonne de prédicat, cette table est entièrement analysée à chaque refresh.

Diagnostiquer le fallback au calcul complet

Lorsqu’un refresh revient à un recalcul complet, la raison est indiquée dans l’événement planning_information du flux. Voir l’objet IssueType dans le schéma du journal des événements du pipeline. Le tableau suivant liste les raisons indiquées dans l’événement :

Raison	Signification
`EXTERNAL_CHANGE_IN_REPLACE_WINDOW`	Une instruction DML externe a modifié des lignes dans la fenêtre de remplacement actuelle.
`REPLACE_WHERE_NOT_DETERMINISTIC`	Le prédicat utilise des expressions non déterministes.
`PRIOR_REPLACE_WHERE_NOT_DETERMINISTIC`	Le refresh précédent a utilisé un prédicat non déterministe.
`PRIOR_REPLACE_WHERE_PREDICATE_MISSING`	Le prédicat utilisé dans la mise à jour précédente n’a pas pu être récupéré. Cela peut se produire si le prédicat n’a pas été sérialisé correctement lors de cette mise à jour.
`STALE_TARGET_DATA_IN_WINDOW`	Le prédicat a été modifié entre les mises à jour et la fenêtre actuelle comprend les lignes que le prédicat précédent excluait.
`CANNOT_PUSH_PREDICATE`	Le prédicat ne peut pas être transféré vers au moins une source.
`REPLACE_WHERE_BACKFILL`	La mise à jour est un remplissage qui utilise une substitution de prédicat unique pour remplir les données historiques ou corriger les écritures antérieures.

Raison	Signification
`EXTERNAL_CHANGE_IN_REPLACE_WINDOW`	Une instruction DML externe a modifié des lignes dans la fenêtre de remplacement actuelle.
`REPLACE_WHERE_NOT_DETERMINISTIC`	Le prédicat utilise des expressions non déterministes.
`PRIOR_REPLACE_WHERE_NOT_DETERMINISTIC`	Le refresh précédent a utilisé un prédicat non déterministe.
`PRIOR_REPLACE_WHERE_PREDICATE_MISSING`	Le prédicat utilisé dans la mise à jour précédente n’a pas pu être récupéré. Cela peut se produire si le prédicat n’a pas été sérialisé correctement lors de cette mise à jour.
`STALE_TARGET_DATA_IN_WINDOW`	Le prédicat a été modifié entre les mises à jour et la fenêtre actuelle comprend les lignes que le prédicat précédent excluait.
`CANNOT_PUSH_PREDICATE`	Le prédicat ne peut pas être transféré vers au moins une source.
`REPLACE_WHERE_BACKFILL`	La mise à jour est un remplissage qui utilise une substitution de prédicat unique pour remplir les données historiques ou corriger les écritures antérieures.

Exemples

L'exemple suivant montre des modèles courants de flux REPLACE WHERE.

Exemple 1 : conserver des agrégats historiques provenant d'une source à conservation limitée

Cet exemple conserve indéfiniment des agrégats quotidiens, même après que les données brutes ont disparu de la table source (conservation de 3 jours) :

SQL
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;

Exemple 2 : Empêcher le recalcul en cas de modification d’une table de dimension

Cet exemple conserve les lignes de faits historiques inchangées lorsque les attributs de dimension changent :

SQL
CREATE OR REFRESH STREAMING TABLE fact_dim_join
FLOW REPLACE WHERE 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 la région d’un utilisateur change, seules les lignes récentes sont recalculées. Les lignes historiques conservent la valeur de la région au moment de leur écriture.

Exemple 3 : ajouter une nouvelle métrique sans recalculer tout l'historique

Cet exemple montre comment faire évoluer une définition de table et effectuer un remplissage uniquement sur une plage ciblée :

Définir la table initiale :

SQL
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;

Mettre à jour la query pour ajouter uniq_users:

SQL
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;

Les lignes plus anciennes que la fenêtre de 7 jours contiennent NULL pour uniq_users.

Exemple 4 : Itérer sur une petite fenêtre avant de remplir l'historique complet

Cet exemple montre comment valider la logique de query sur une petite fenêtre de données avant de traiter l'ensemble de la plage historique.

start avec une fenêtre courte pour valider les métriques et itérer sur la logique métier avec des coûts de compute inférieurs :

SQL
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;

Une fenêtre courte recalcule uniquement les 7 derniers jours à chaque refresh. Revoyez donc la query autant de fois que nécessaire avant de vous engager dans un commit historique complet.

Une fois la query finalisée, utilisez le DML pour remplir la plage historique complète :

SQL
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;

Configuration requise​

Quand utiliser les flux REPLACE WHERE​

Créez un flux REPLACE WHERE​

Remplissage rétroactif des données historiques​

Remplacements de prédicats​

Instructions DML​

Comportement de refresh complète​

Actualisation refresh​

Quand l'refresh incrémentiel s'applique​

Meilleures pratiques pour l’refresh incrémentiel​

Utiliser une limite inférieure mobile​

Inclure la colonne prédicat dans GROUP BY​

Inclure la colonne de prédicat dans les clés de jointure​

Diagnostiquer le fallback au calcul complet​

Exemples​

Exemple 1 : conserver des agrégats historiques provenant d'une source à conservation limitée​

Exemple 2 : Empêcher le recalcul en cas de modification d’une table de dimension​

Exemple 3 : ajouter une nouvelle métrique sans recalculer tout l'historique​

Exemple 4 : Itérer sur une petite fenêtre avant de remplir l'historique complet​