Caricare dati usando tabelle di streaming in Databricks SQL

Databricks consiglia di usare le tabelle di streaming per inserire dati usando Databricks SQL. Una tabella di streaming è una tabella registrata in Unity Catalog con supporto aggiuntivo per lo streaming o l'elaborazione incrementale dei dati. Per ogni tabella di streaming viene creata automaticamente una pipeline DLT. È possibile usare le tabelle di streaming per il caricamento incrementale dei dati da Kafka e dall'archiviazione di oggetti cloud.

Questo articolo illustra l'uso delle tabelle di streaming per caricare i dati dall'archiviazione di oggetti cloud configurata come volume del catalogo Unity (scelta consigliata) o come posizione esterna.

Prima di iniziare

Prima di iniziare, è necessario soddisfare i requisiti seguenti:

Requisiti dell'area di lavoro:

• Un account Azure Databricks con l'opzione serverless abilitata. Per altre informazioni, vedere Abilitare SQL Warehouse serverless.
• Un'area di lavoro con il catalogo Unity abilitato. Per altre informazioni, vedere Configurare e gestire Il catalogo Unity.

Requisiti di calcolo:

È necessario usare uno dei seguenti elementi:

• Un magazzino SQL che utilizza il canale Current.
• Calcolo con modalità di accesso standard (in precedenza modalità di accesso condiviso) in Databricks Runtime 13.3 LTS o versione successiva.

• Calcolo con modalità di accesso dedicato (in precedenza modalità di accesso utente singolo) in Databricks Runtime 15.4 LTS o versione successiva.

  In Databricks Runtime 15.3 e versioni successive non è possibile usare risorse di calcolo dedicate per eseguire query sulle tabelle di streaming di proprietà di altri utenti. È possibile usare calcolo dedicato in Databricks Runtime 15.3 e versioni successive solo se si è proprietari della tabella di streaming. Il creatore della tabella è il proprietario.

  Databricks Runtime 15.4 LTS e versioni successive supportano query su tabelle generate da DLT in calcolo dedicato, indipendentemente dalla proprietà della tabella. Per sfruttare i vantaggi del filtro dei dati fornito in Databricks Runtime 15.4 LTS e versioni successive, è necessario verificare che 'area di lavoro sia abilitata per l'elaborazione serverless perché la funzionalità di filtro dei dati che supporta le tabelle generate da DLT viene eseguita nel calcolo serverless. È possibile che vengano addebitati costi per le risorse di calcolo serverless quando si usa un calcolo dedicato per eseguire operazioni di filtro dei dati. Vedere controllo di accesso con granularità fine in un ambiente di calcolo dedicato (in precedenza calcolo utente singolo).

Requisiti di autorizzazione:

• Il privilegio READ FILES relativo a una posizione esterna di Unity Catalog. Per informazioni, vedere Creare una posizione esterna per connettere la memorizzazione cloud ad Azure Databricks.
• Il privilegio USE CATALOG sul catalogo in cui si crea la tabella di streaming.
• Privilegio USE SCHEMA sullo schema in cui si crea la tabella di streaming.
• Privilegio CREATE TABLE sullo schema in cui si crea la tabella di streaming.

Altri requisiti:

• Il percorso ai tuoi dati di origine.

  Esempio di percorso del volume: /Volumes/<catalog>/<schema>/<volume>/<path>/<file-name>

  Esempio di percorso esterno: abfss://myContainer@myStorageAccount.dfs.core.windows.net/analysis

  Nota

  Questo articolo presuppone che i dati da caricare si trovino in un percorso di archiviazione cloud che corrisponde a un volume del catalogo Unity o a un percorso esterno a cui si ha accesso.

Individuare e visualizzare in anteprima i dati di origine

1. Nella barra laterale dell'area di lavoro cliccare su Query e quindi su Crea query.

2. Nell'editor di query, selezionare un SQL warehouse che utilizza il canale Current dall'elenco a discesa.

3. Incolla il codice seguente nell'editor, sostituendo i valori tra parentesi angolari (<>) con le informazioni che identificano i dati di origine, e quindi fai clic su Esegui.

   Nota

   È possibile che si verifichino errori di inferenza dello schema durante l'esecuzione della funzione con valori di tabella read_files se le impostazioni predefinite per la funzione non possono analizzare i dati. Ad esempio, potrebbe essere necessario configurare la modalità a più righe per i file CSV o JSON su più righe. Per un elenco delle opzioni del parser, vedere read_files funzione con valori di tabella.

   /* Discover your data in a volume */
   LIST "/Volumes/<catalog>/<schema>/<volume>/<path>/<folder>"
   
   /* Preview your data in a volume */
   SELECT * FROM read_files("/Volumes/<catalog>/<schema>/<volume>/<path>/<folder>") LIMIT 10
   
   /* Discover your data in an external location */
   LIST "abfss://<container>@<storage-account>.dfs.core.windows.net/<path>/<folder>"
   
   /* Preview your data */
   SELECT * FROM read_files("abfss://<container>@<storage-account>.dfs.core.windows.net/<path>/<folder>") LIMIT 10
   

Caricare dati in una tabella di streaming

Per creare una tabella di streaming dai dati nell'archivio di oggetti cloud, incollare quanto segue nell'editor di query e quindi fare clic su Esegui:

/* Load data from a volume */
CREATE OR REFRESH STREAMING TABLE <table-name> AS
SELECT * FROM STREAM read_files('/Volumes/<catalog>/<schema>/<volume>/<path>/<folder>')

/* Load data from an external location */
CREATE OR REFRESH STREAMING TABLE <table-name> AS
SELECT * FROM STREAM read_files('abfss://<container>@<storage-account>.dfs.core.windows.net/<path>/<folder>')

Impostare il canale di esecuzione

Le tabelle di streaming create con sql warehouse vengono aggiornate automaticamente tramite una pipeline DLT. Le pipeline DLT usano il runtime nel canale current per impostazione predefinita. Per informazioni sul processo di rilascio, vedere note sulla versione DLT e il processo di aggiornamento della versione.

Databricks consiglia di usare il current canale per i carichi di lavoro di produzione. Le nuove funzionalità vengono rilasciate per la prima volta al preview canale. È possibile impostare una pipeline sul canale DLT di anteprima per testare le nuove funzionalità specificando preview come proprietà della tabella. È possibile specificare questa proprietà quando si crea la tabella o dopo la creazione della tabella utilizzando un'istruzione ALTER.

L'esempio di codice seguente illustra come impostare il canale per l'anteprima in un'istruzione CREATE:

CREATE OR REPLACE MATERIALIZED VIEW foo.default.bar
TBLPROPERTIES ('pipelines.channel' = 'preview') as
SELECT
  *
FROM
  range(5)

Aggiornare una tabella di streaming usando una pipeline DLT

Questa sezione descrive i modelli per aggiornare una tabella di streaming con i dati più recenti disponibili dalle origini definite nella query.

Quando esegui CREATE o REFRESH su una tabella di streaming, l'elaborazione dell'aggiornamento avviene usando una pipeline DLT serverless. Ogni tabella di streaming definita ha una pipeline DLT associata.

Dopo aver eseguito il comando REFRESH, viene restituito il collegamento alla pipeline DLT. È possibile usare il collegamento della pipeline DLT per controllare lo stato dell'aggiornamento.

Vedere Che cos'è DLT?.

Inserire solo nuovi dati

Per impostazione predefinita, la funzione read_files legge tutti i dati esistenti nella directory di origine durante la creazione della tabella e quindi elabora i record appena arrivati con ogni aggiornamento.

Per evitare di inserire dati già presenti nella directory di origine al momento della creazione della tabella, impostare l'opzione includeExistingFiles su false. Ciò significa che vengono elaborati solo i dati che arrivano nella directory dopo la creazione della tabella. Ad esempio:

CREATE OR REFRESH STREAMING TABLE my_bronze_table
AS SELECT *
FROM STREAM read_files(
  'abfss://myContainer@myStorageAccount.dfs.core.windows.net/analysis/*/*/*.json',
  includeExistingFiles => false)

Aggiornare completamente una tabella di streaming

L'aggiornamento completo rielabora tutti i dati disponibili nell'origine usando la definizione più recente. Non è consigliabile chiamare aggiornamenti completi sulle origini che non mantengono l'intera cronologia dei dati o hanno brevi periodi di conservazione, ad esempio Kafka, perché l'aggiornamento completo tronca i dati esistenti. Potrebbe non essere possibile recuperare i dati obsoleti se i dati non sono più disponibili nell'origine.

Ad esempio:

REFRESH STREAMING TABLE my_bronze_table FULL

Pianificare una tabella di streaming per l'aggiornamento automatico

Per configurare una tabella di streaming per l'aggiornamento automatico in base a una pianificazione definita, incollare quanto segue nell'editor di query e quindi fare clic su Esegui:

ALTER STREAMING TABLE
[[<catalog>.]<database>.]<name>
ADD [SCHEDULE [REFRESH]
        CRON '<cron-string>'
                [ AT TIME ZONE '<timezone-id>' ]];

Per esempio, per le query di aggiornamento pianificate, vedere ALTER STREAMING TABLE.

Tenere traccia dello stato di un aggiornamento

È possibile visualizzare lo stato di un aggiornamento della tabella di streaming visualizzando la pipeline che gestisce la tabella di streaming nell'interfaccia utente DLT o visualizzando le informazioni di aggiornamento restituite dal comando DESCRIBE EXTENDED per la tabella di streaming.

DESCRIBE EXTENDED <table-name>

Inserimento in streaming da Kafka

Per un esempio di inserimento in streaming da Kafka, vedere read_kafka.

Concedere agli utenti l'accesso a una tabella di streaming

Per concedere agli utenti il privilegio di SELECT nella tabella di streaming in modo che possano eseguire query, incollare quanto segue nell'editor di query e quindi fare clic su Esegui:

GRANT SELECT ON TABLE <catalog>.<schema>.<table> TO <user-or-group>

Per altre informazioni sulla concessione dei privilegi per gli oggetti a protezione diretta di Unity Catalog, vedere i privilegi del catalogo Unity e gli oggetti a protezione diretta.

eliminare definitivamente i record da una tabella di streaming

Per eliminare fisicamente i record dall'archiviazione sottostante per una tabella di streaming con vettori di eliminazione abilitati, ad esempio per la conformità al GDPR, è necessario eseguire ulteriori passaggi per garantire che un'operazione di VACUUM venga eseguita sui dati della tabella di streaming.

Di seguito vengono descritti in modo più dettagliato questi passaggi:

1. Aggiornare i record o eliminare record dalla tabella di streaming.
2. Eseguire un'istruzione REORG sulla tabella di streaming, specificando il parametro APPLY (PURGE). Ad esempio, REORG TABLE <streaming-table-name> APPLY (PURGE);.
3. Attendere il termine del periodo di conservazione dei dati della tabella di streaming. Il periodo di conservazione dei dati predefinito è di sette giorni, ma può essere configurato con la proprietà della tabella delta.deletedFileRetentionDuration. Vedere Configurare la conservazione dei dati per le query di spostamento cronologico.
4. REFRESH la tabella di streaming. Vedere Aggiornare una tabella di streaming usando una pipeline DLT. Entro 24 ore dall'operazione di REFRESH, le attività di manutenzione DLT, inclusa l'operazione di VACUUM necessaria per garantire che i record vengano eliminati definitivamente, vengono eseguiti automaticamente. Vedere Attività di manutenzione eseguite da DLT.

Monitorare le esecuzioni usando la cronologia delle query

È possibile usare la pagina cronologia query per accedere ai dettagli delle query e ai profili di query che consentono di identificare query con prestazioni scarse e colli di bottiglia nella pipeline DLT usata per eseguire gli aggiornamenti delle tabelle di streaming. Per una panoramica del tipo di informazioni disponibili nelle cronologie delle query e nei profili di query, vedere Cronologia query e Profilo di query.

Tutte le dichiarazioni relative alle tabelle di streaming appaiono nella cronologia delle query. È possibile utilizzare il filtro a discesa Statement per selezionare qualsiasi comando ed esaminare le query correlate. Tutte le istruzioni CREATE sono seguite da un'istruzione REFRESH eseguita in modo asincrono in una pipeline DLT. Le REFRESH istruzioni in genere includono piani di query dettagliati che forniscono informazioni sull'ottimizzazione delle prestazioni.

Per accedere alle istruzioni REFRESH nella cronologia delle query dell'interfaccia utente, seguire questi passaggi:

1. Clicca su nella barra laterale sinistra per aprire l'interfaccia Cronologia Query.
2. Selezionare la casella di controllo REFRESH dal filtro a discesa Dichiarazione.
3. Fare clic sul nome dell'istruzione query per visualizzare i dettagli di riepilogo, ad esempio la durata della query e le metriche aggregate.
4. Fare clic su Visualizza profilo di query per aprire il profilo di query. Per informazioni dettagliate sull'esplorazione del profilo di query, vedere Profilo di query .
5. Facoltativamente, è possibile usare i collegamenti nella sezione Query di origine per aprire la query o la pipeline correlata.

È anche possibile accedere ai dettagli delle query usando collegamenti nell'editor SQL o da un notebook collegato a un'istanza di SQL Warehouse.

Risorse aggiuntive

• tabella di streaming
• read_files funzione con valori di tabella
• CREATE STREAMING TABLE
• ALTER STREAMING TABLE
• read_kafka funzione con valori di tabella