CRÉER UN FLUX (pipelines)

Utilisez l’instruction CREATE FLOW pour créer des flux ou des remplissages pour les tables d’un pipeline.

Note

CREATE FLOW Le ciblage d’une table de streaming prend en charge à la fois AUTO CDC ... INTO les flux et REPLACE WHERE et . Une table managée créée avec CREATE TABLE ... FLOW ne supporte pas la capture de données de changement : un AUTO CDC ... INTO flux contre une table gérée échoue avec MANAGED_TABLE_DOES_NOT_SUPPORT_CDC. CREATE FLOW Into a Streaming Table n’est pas soumis à cette limitation. Voir CREATE TABLE ... FLOW (pipelines).

Syntaxe

CREATE FLOW flow_name [COMMENT comment] AS
{
  AUTO CDC [ONCE] INTO target_table create_auto_cdc_flow_spec |
  AUTO CDC [ONCE] INTO target_table create_auto_cdc_from_snapshot_spec |
  INSERT [ONCE] INTO target_table BY NAME [ replace_using_spec | REPLACE WHERE condition ] query
}

create_auto_cdc_from_snapshot_spec
  FROM SNAPSHOT ( snapshot_query )
  [ WITH VERSION ( version_query ) ]
  KEYS ( key [, ...] )
  [ STORED AS { SCD TYPE 1 | SCD TYPE 2 } ]
  [ TRACK HISTORY ON { col_list | * EXCEPT ( col_list ) } ]

replace_using_spec
  REPLACE USING ( column_name [, ...] ) SEQUENCE BY sequence_column

Paramètres

  • flow_name

    Nom du flux à créer.

  • COMMENTAIRE

    Description facultative du flux.

  • AUTO-CDC EN

    Instruction AUTO CDC ... INTO qui définit le flux, avec un create_auto_cdc_flow_spec. Vous devez inclure une AUTO CDC ... INTO instruction ou une INSERT INTO instruction. Utilisez AUTO CDC ... INTO quand la requête source utilise la sémantique des données modifiées.

    Pour plus d’informations, consultez AUTO CDC INTO (pipelines).

  • AUTO CDC ... EXTRAIT DE SNAPSHOT

    Une AUTO CDC ... INTO instruction qui dérive des changements en comparant des instantanés au lieu de lire un flux de changement. Utilisez ce formulaire lorsque la capture de données de modification n’est pas activée sur la source et que seuls des instantanés complets sont disponibles. La source est spécifiée en deux parties : une clause obligatoire FROM SNAPSHOT (snapshot_query) qui lit les données instantanées, et une clause optionnelle WITH VERSION (version_query) qui sélectionne la prochaine version de l’instantané à traiter. Découvrez comment fonctionne AUTO CDC FROM SNAPSHOT.

    • EXTRAIT DE L’INSTANTANÉ (snapshot_query)

      Required. Une requête qui lit les données instantanées pour la version sélectionnée par WITH VERSION (...). Le moteur différencie le résultat de l’instantané précédemment engagé pour dériver les insertions, mises à jour et suppressions, et les fusionne dans la cible en utilisant KEYS l’identité de ligne et STORED AS pour déterminer comment les changements sont stockés.

      Appelez current_snapshot_version() à l’intérieur de cette requête pour référencer la version sélectionnée par WITH VERSION (...). Si WITH VERSION (...) n’est pas spécifié, current_snapshot_version() n’est pas appelable à l’intérieur FROM SNAPSHOT (...)de .

      Lorsque WITH VERSION (...) la cible est omise, le moteur lit directement la source via FROM SNAPSHOT (...), et la requête instantanée ne s’exécute que pendant le chargement initial, tandis que la cible n’a ni données engagées ni état d’instantané engagé. Lors de toute mise à jour ultérieure, lorsque la cible contient déjà des données ou a confirmé l’état d’instantané, le flux échoue avec AUTO_CDC_FROM_SNAPSHOT_NON_EMPTY_TARGET_WITHOUT_VERSION. Pour traiter les instantanés sur plusieurs mises à jour, utilisez WITH VERSION (...).

    • AVEC VERSION (version_query)

      Optional. Une requête qui sélectionne la prochaine version snapshot à traiter. Il doit retourner exactement une colonne de type ordonné, soit 0 ou 1 ligne. Lorsqu’il retourne 1 ligne, la valeur doit être non nulle. La colonne peut être une valeur scalaire, comme un BIGINT, ou un STRUCT dont les champs sont tous ordonnés. Une requête de version qui retourne plus d’une colonne, plus d’une ligne ou une valeur nulle échoue au flux avec INVALID_AUTO_CDC_FROM_SNAPSHOT_VERSION_QUERY.

      Lors d’une mise à jour du pipeline, le moteur répète les étapes suivantes : il évalue la requête de version ; si la requête ne retourne aucune ligne, elle arrête de traiter ce flux pour la mise à jour en cours ; si la requête retourne 1 ligne, le moteur expose cette valeur via current_snapshot_version(), évalue la requête instantanée, envoie le snapshot résultant, et expose la version confirmée via last_snapshot_version(). Le moteur réévalue ensuite la requête de version pour sélectionner la version suivante. Une seule mise à jour de pipeline traite les versions dans l’ordre jusqu’à ce que la requête de version ne renvoie aucune ligne.

      Chaque version retournée après un commit réussi doit être supérieure à la version précédemment validée ; une version non croissante échoue à la mise à jour avec APPLY_CHANGES_FROM_SNAPSHOT_ERROR.OUT_OF_ORDER_SNAPSHOT_VERSION. Le type de donnée de la valeur de version doit rester inchangé à travers les commits d’instantané ; un changement de type de données échoue à la mise à jour avec AUTO_CDC_FROM_SNAPSHOT_VERSION_SCHEMA_CHANGED. Un rafraîchissement complet efface l’état persistant de la version.

    • CLÉS

      Required. Les colonnes clés principales servent à identifier les lignes à travers les instantanés pour la détection des changements.

    • STORED AS { SCD TYPE 1 | SCD TYPE 2 }

      Optional. Spécifie comment les changements sont stockés dans la table cible. La valeur par défaut est SCD TYPE 1.

    • TRACK HISTORY ON { col_list | * EXCEPT (col_list) }

      Optional. S’applique uniquement avec SCD TYPE 2. Précise quelles colonnes déclenchent une nouvelle ligne d’historique lorsqu’elles changent. Fournir soit une liste * EXCEPT (col_list) de colonnes explicite, soit suivre toutes les colonnes sauf celles listées.

    Snapshot CDC ne prend pas en charge WHERE ni SEQUENCE BY. L’ordre croisé des instantanés s’exprime par WITH VERSION (...).

  • target_table

    Table à mettre à jour. Il doit s’agir d’une table de streaming.

  • INSERT DANS

    Définit une requête de table insérée dans la table cible. Si l’option ONCE n’est pas fournie, la requête doit être une requête de diffusion en continu . Utilisez le mot clé STREAM pour utiliser la sémantique de diffusion en continu pour lire à partir de la source. Si la lecture détecte une modification ou une suppression concernant un enregistrement existant, une erreur est générée. Il est plus sûr de lire depuis des sources statiques ou d’ajout uniquement. Pour ingérer des données ayant des validations de modification, vous pouvez utiliser Python et l’option skipChangeCommits permettant de gérer les erreurs.

    INSERT INTO s’exclue mutuellement avec AUTO CDC ... INTO. Utiliser AUTO CDC ... INTO lorsque les données sources incluent des fonctionnalités de capture de données modifiées (CDC). Utilisez INSERT INTO quand la source ne le fait pas.

    Pour plus d’informations sur la diffusion en continu des données, consultez Transformer des données avec des pipelines.

  • REMPLACER PAR ( column_name [, ...] ) SÉQUENCE PAR sequence_column

    Important

    Cette fonctionnalité est en version bêta. Nécessite Databricks Runtime 18.2 et plus.

    Définit le flux comme un REPLACE USING flux, qui remplace toutes les lignes de la table cible correspondant aux colonnes clés spécifiées et laisse toutes les autres lignes intactes. Utilisez REPLACE USING lorsque votre source est une série d’instantanés partiels codés par colonne. SEQUENCE BY ordonne les mises à jour pour que la séquence la plus élevée pour une clé l’emporte, même lorsque les mises à jour arrivent dans le désordre.

    Spécifiez au moins une colonne clé et exactement une SEQUENCE BY colonne. La requête doit être une requête en streaming, et BY NAME elle est obligatoire. REPLACE USING ne peut pas être combiné avec ONCE ou avec AUTO CDC ... INTO.

    Pour plus d’informations, voir Remplacement partiel d’instantané par REMPLACER les flux UTILISÉS.

  • Condition REMPLACER WHERE

    Définit le flux comme un REPLACE WHERE flux, qui recalcule et écrase un sous-ensemble ciblé de la table cible. À chaque mise à jour, toutes les lignes de la correspondance condition de la table cible sont supprimées, la requête source est recalculée pour cette même plage de prédicats, et les résultats sont insérés. Les rangées qui ne correspondent condition pas restent intactes. Vous n’avez pas besoin d’ajouter le prédicat à la requête source ; le moteur pipeline l’applique automatiquement lors de la lecture de la source.

    REPLACE WHERE Utilise la sémantique batch, donc la requête source n’a pas besoin d’être une requête en streaming. BY NAME est obligatoire. REPLACE WHERE ne peut pas être combiné avec ONCE, REPLACE USINGou AUTO CDC ... INTO.

    Pour plus d’informations, voir Traitement par lots avec flux REPLACE WHERE et flux REPLACE WHERE pour les tables de flux autonomes.

  • ONCE

    Si vous le souhaitez, définissez le flux comme un flux à usage unique, tel qu’un remblai. L'utilisation de ONCE change le flux de deux manières :

    • La source query ou create_auto_cdc_flow_spec n’est pas une table de flux.
    • Le flux est exécuté une fois par défaut. Si le pipeline est mis à jour avec une actualisation complète, le ONCE flux s’exécute à nouveau pour recréer les données.

    ONCE ne peut pas être utilisé avec REPLACE USING, ce qui nécessite une source de streaming.

Examples

-- EXAMPLE 1:
-- Create a streaming table, and add two flows that append data to it:
CREATE OR REFRESH STREAMING TABLE users;

-- first flow into target_table:
CREATE FLOW users_flow AS
INSERT INTO users BY NAME
SELECT * FROM stream(raw_data.users);

-- second flow into target_table:
CREATE FLOW backfill_users AS
INSERT ONCE INTO users BY NAME
SELECT * FROM user_backfill_table;

-- EXAMPLE 2:
-- Create a streaming table, and add a flow that applies CDC changes to it:
CREATE OR REFRESH STREAMING TABLE admins_cdc_target_table;

-- first flow into target_table:
CREATE FLOW admin_cdc_flow AS
AUTO CDC INTO admins_cdc_target_table
FROM stream(cdc_data.admins)
KEYS (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY sequenceNum
COLUMNS * EXCEPT (operation, sequenceNum)
STORED AS SCD TYPE 2;

-- EXAMPLE 3:
-- Create a streaming table, and add a REPLACE USING flow that keeps the latest
-- row for each payment_id from a stream of partial snapshots:
CREATE OR REFRESH STREAMING TABLE payments_latest;

CREATE FLOW payments_replace_flow AS
INSERT INTO payments_latest BY NAME
REPLACE USING (payment_id) SEQUENCE BY payment_date
SELECT payment_id, booking_id, status, payment_date
FROM STREAM(samples.wanderbricks.payments);

-- EXAMPLE 4:
-- AUTO CDC FROM SNAPSHOT without WITH VERSION: a one-time initial load from a snapshot table.
-- To process later snapshots on each update, add WITH VERSION (see EXAMPLE 5).
CREATE STREAMING TABLE users (user_id INT, name STRING, email STRING);

CREATE FLOW users_snapshot_flow AS
AUTO CDC ONCE INTO users
FROM SNAPSHOT (SELECT * FROM catalog.schema.users_snapshot)
KEYS (user_id)
STORED AS SCD TYPE 1;

-- EXAMPLE 5:
-- AUTO CDC FROM SNAPSHOT with WITH VERSION: pick the next file, then read it as the snapshot:
CREATE STREAMING TABLE orders (order_id INT, product STRING, quantity INT, order_date DATE);

CREATE FLOW orders_cdc AS
AUTO CDC INTO orders
FROM SNAPSHOT (
  SELECT order_id, product, quantity, order_date
  FROM read_files('/Volumes/catalog/schema/landing/orders/', format => 'json')
  WHERE _metadata.file_path = (SELECT version.path FROM current_snapshot_version())
)
WITH VERSION (
  SELECT struct(modification_time, path) AS version
  FROM list_files('/Volumes/catalog/schema/landing/orders/')
  WHERE (
    NOT EXISTS (SELECT 1 FROM last_snapshot_version())
    OR struct(modification_time, path) > (SELECT version FROM last_snapshot_version())
  )
  ORDER BY modification_time, path
  LIMIT 1
)
KEYS (order_id)
STORED AS SCD TYPE 2;

-- EXAMPLE 6:
-- One-time snapshot backfill plus a streaming CDC flow into the same target.
-- The backfill omits WITH VERSION, so it uses an implicit timestamp version. The
-- streaming flow's SEQUENCE BY column (event_ts) must be a TIMESTAMP so its type
-- matches that implicit version on the shared target.
CREATE STREAMING TABLE customers (
  customer_id INT, name STRING, email STRING, address STRING, event_ts TIMESTAMP
);

CREATE FLOW customers_snapshot_backfill AS
AUTO CDC ONCE INTO customers
FROM SNAPSHOT (SELECT * FROM catalog.schema.customers_snapshot)
KEYS (customer_id)
STORED AS SCD TYPE 1;

CREATE FLOW customers_cdc AS
AUTO CDC INTO customers
FROM STREAM(customers_cdc_events)
KEYS (customer_id)
SEQUENCE BY event_ts
STORED AS SCD TYPE 1;

-- EXAMPLE 7:
-- Create a streaming table, then add a REPLACE WHERE flow that recomputes and
-- overwrites a targeted window of the target table on each update:
CREATE STREAMING TABLE payments_latest;

CREATE FLOW payments_latest AS
INSERT INTO payments_latest BY NAME
REPLACE WHERE payment_date >= date_add(current_date(), -7)
SELECT payment_id, booking_id, status, payment_date
FROM samples.wanderbricks.payments;

Pour en savoir plus sur la combinaison d’un remblai ponctuel avec un CDC en cours sur la même cible, voir Remblai des données historiques avec pipelines.

  • Dernière mise à jour :