CREA FLUSSO (pipeline)

Usare l'istruzione CREATE FLOW per creare flussi o backfill per le tabelle in una pipeline.

Sintassi

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

Parametri

• flow_name

  Nome del flusso da creare.

• COMMENTO

  Descrizione facoltativa per il flusso.

• AUTO CDC IN

  Istruzione AUTO CDC ... INTO che definisce il flusso, con un elemento create_auto_cdc_flow_spec. È necessario includere un'istruzione AUTO CDC ... INTO o un'istruzione INSERT INTO . Usare AUTO CDC ... INTO quando la query di origine usa la semantica delle modifiche ai dati.

  Per altre informazioni, vedere AUTO CDC INTO (pipelines).

• AUTO CDC ... DA SNAPSHOT

  Un'affermazione AUTO CDC ... INTO che deriva le modifiche confrontando istantanee invece di leggere un feed di modifica. Utilizzare questo modulo quando la raccolta dei dati di modifica non è abilitata sulla sorgente e sono disponibili solo istantanee complete. La sorgente è specificata in due parti: una clausola obbligatoria FROM SNAPSHOT (snapshot_query) che legge i dati snapshot e una clausola opzionale WITH VERSION (version_query) che seleziona la prossima versione snapshot da elaborare. Vedere Funzionamento di AUTO CDC FROM SNAPSHOT.

  • DA ISTANTANEO (snapshot_query)

    Required. Una query che legge i dati snapshot per la versione selezionata da WITH VERSION (...). Il motore differenzia il risultato dallo snapshot precedentemente valido per derivare inserti, aggiornamenti e cancellazioni, e li unisce nella destinazione usando KEYS l'identità della riga e STORED AS per determinare come le modifiche vengono memorizzate.

    Chiama current_snapshot_version() all'interno di questa query per fare riferimento alla versione selezionata da WITH VERSION (...). Se WITH VERSION (...) non è specificato, current_snapshot_version() non è chiamabile all'interno FROM SNAPSHOT (...)di .

    Quando WITH VERSION (...) viene omesso, il motore legge direttamente la sorgente attraverso FROM SNAPSHOT (...), e la query snapshot viene eseguita solo durante il caricamento iniziale, mentre il target non ha dati confermati né stato snapshot commesso. In qualsiasi aggiornamento successivo, quando il target contiene già dati o ha lo stato snapshot confermato, il flusso fallisce con AUTO_CDC_FROM_SNAPSHOT_NON_EMPTY_TARGET_WITHOUT_VERSION. Per elaborare istantanee su più aggiornamenti, usa WITH VERSION (...).

  • CON VERSIONE (version_query)

    Optional. Una query che seleziona la prossima versione snapshot da elaborare. Deve restituire esattamente una colonna di tipo ordinabile e 0 o 1 riga. Quando restituisce 1 riga, il valore deve essere non nullo. La colonna può essere un valore scalare, come un BIGINT, oppure un STRUCT i cui campi sono tutti ordinabili. Una query di versione che restituisce più di una colonna, più di una riga o un valore nullo fallisce il flusso con INVALID_AUTO_CDC_FROM_SNAPSHOT_VERSION_QUERY.

    Durante un aggiornamento della pipeline, il motore ripete i seguenti passaggi: valuta la query di versione; se la query restituisce 0 righe, interrompe l'elaborazione di questo flusso per l'aggiornamento corrente; se la query restituisce 1 riga, il motore espone quel valore tramite current_snapshot_version(), valuta la query snapshot, effettua un commit per lo snapshot risultante ed espone la versione commessa tramite last_snapshot_version(). Il motore poi rivaluta la query di versione per selezionare la versione successiva. Un singolo aggiornamento della pipeline elabora le versioni in ordine finché la query di versione non restituisce righe.

    Ogni versione restituita dopo un commit riuscito deve essere maggiore della versione precedentemente commessa; una versione non crescente fallisce l'aggiornamento con APPLY_CHANGES_FROM_SNAPSHOT_ERROR.OUT_OF_ORDER_SNAPSHOT_VERSION. Il tipo di dato del valore della versione deve rimanere invariato tra i commit snapshot; una modifica del tipo di dato fallisce l'aggiornamento con AUTO_CDC_FROM_SNAPSHOT_VERSION_SCHEMA_CHANGED. Un aggiornamento completo cancella lo stato persistente della versione.

  • CHAVI

    Required. Le colonne chiave principali sono usate per identificare le righe tra istantanee per il rilevamento delle modifiche.

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

    Optional. Specifica come le modifiche sono memorizzate nella tabella target. Il valore predefinito è SCD TYPE 1.

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

    Optional. Si applica solo con SCD TYPE 2. Specifica quali colonne attivano una nuova riga di cronologia quando cambiano. Fornire una lista esplicita di colonne o * EXCEPT (col_list) tracciare ogni colonna tranne quelle elencate.

  Snapshot CDC non supporta WHERE né SEQUENCE BY. L'ordinamento cross-snapshot è espresso tramite WITH VERSION (...).

• target_table

  Tabella da aggiornare. Deve trattarsi di una tabella di streaming.

• INSERT IN

  Definisce una query di tabella che viene inserita nella tabella di destinazione. Se l'opzione ONCE non viene specificata, la query deve essere una query di streaming . Utilizzare la parola chiave STREAM per utilizzare la semantica di streaming per leggere dalla sorgente. Se la lettura rileva una modifica o un'eliminazione in un record esistente, viene generato un errore. È più sicuro leggere da fonti statiche o a solo aggiunta. Per inserire dati con commit delle modifiche, è possibile usare Python e l'opzione skipChangeCommits per gestire gli errori.

  INSERT INTO è mutuamente esclusivo rispetto a AUTO CDC ... INTO. Usare AUTO CDC ... INTO quando i dati di origine includono la funzionalità Change Data Capture (CDC). Usare INSERT INTO quando la sorgente non lo fa.

  Per ulteriori informazioni sui dati di streaming, vedere Trasformare i dati con le pipeline.

• SOSTITUIRE USANDO ( column_name [, ...] ) SEQUENZA PER sequence_column

  Importante

  Questa funzionalità è in versione beta. Richiede Databricks runtime 18.2 e superiori.

  Definisce il flusso come un REPLACE USING flusso che sostituisce tutte le righe della tabella di destinazione corrispondenti alle colonne chiave specificate e lascia tutte le altre righe intatte. Usalo REPLACE USING quando la tua sorgente è una serie di istantanee parziali chiaviate per colonna. SEQUENCE BY Ordina gli aggiornamenti in modo che vinca la sequenza più alta per una chiave, anche quando gli aggiornamenti arrivano fuori ordine.

  Specifica almeno una colonna chiave e esattamente una SEQUENCE BY colonna. La query deve essere una query in streaming ed BY NAME è obbligatoria. REPLACE USING non può essere combinato con ONCE o con AUTO CDC ... INTO.

  Per maggiori informazioni, vedi Sostituzione parziale di snapshot con SOSTITUIRE i flussi USANTI.

• Condizione REPLACE WHERE

  Definisce il flusso come un REPLACE WHERE flusso che ricalcola e sovrascrive un sottoinsieme target della tabella target. Ad ogni aggiornamento, tutte le righe della tabella di riferimento condition vengono eliminate, la query sorgente viene ricalcolata per lo stesso intervallo di predicati e i risultati vengono inseriti. Le file che non corrispondono condition restano intatte. Non è necessario aggiungere il predicato alla query sorgente; il motore pipeline lo applica automaticamente quando si legge dalla sorgente.

  REPLACE WHERE Usa semantica batch, quindi la query sorgente non deve essere necessariamente una query in streaming. BY NAME è obbligatorio. REPLACE WHERE non può essere combinato con ONCE, REPLACE USING, o AUTO CDC ... INTO.

  Per maggiori informazioni, vedi Elaborazione Batch con flussi REPLACE WHERE e REPLACE WHERE flows per le tabelle di streaming standalone.

• Una volta

  È possibile definire il flusso come un flusso singolo, ad esempio un backfill. L'uso di ONCE modifica il flusso in due modi:

  • L'origine query o create_auto_cdc_flow_spec non è una tabella di streaming.
  • Il flusso viene eseguito una sola volta per impostazione predefinita. Se la pipeline viene aggiornata con un aggiornamento completo, il ONCE flusso viene eseguito di nuovo per ricreare i dati.

  ONCE Non può essere usato con REPLACE USING, che richiede una sorgente di streaming.

Esempi

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

Per maggiori informazioni sulla combinazione di un backfill una tantum con il CDC in corso sullo stesso target, vedi Backfill data historical with pipelines.