Cosa fa effettivamente il destinatario con una tabella condivisa
Molte architetture di condivisione dei dati iniziano con uno schema familiare:
La Parte 1 di questo blog ha descritto perché una tabella condivisa può essere un'alternativa migliore a un file ricorrente o a un feed API. Ora vediamo cosa fa effettivamente il destinatario con quella tabella.
Questa procedura dettagliata si concentra sulla condivisione Databricks-to-Databricks (D2D). Il destinatario dispone di un workspace Databricks abilitato per Unity Catalog e accede alla tabella condivisa tramite un catalogo condiviso. Il fornitore condivide la tabella con la sua cronologia, consentendo al destinatario di usarla come origine Structured Streaming. Torneremo sulla condivisione Databricks-to-Open (D2O) più avanti nell'articolo.
Una Delta table mantiene una cronologia ordinata di ogni modifica apportata. OpenSharing espone tale tabella tramite un protocollo aperto senza copiarla. I dati rimangono nello storage del fornitore e il destinatario li legge in tempo reale.
Dal lato del destinatario, si punta un lettore Structured Streaming alla tabella condivisa. Spark registra fino a che punto ha letto in un checkpoint, quindi ogni esecuzione preleva solo le modifiche apportate dall'ultima. Tenere traccia di "cosa c'è di nuovo dall'ultima volta che ho controllato" diventa il compito del motore, che viene eseguito su dati governati con lineage e controllo degli accessi intatti.
Facciamo un esempio: un venditore al dettaglio gestisce una tabella di ordini in tempo reale e la condivende con un destinatario esterno che esegue Databricks. Invece di ricevere un CSV ogni notte, il destinatario può interrogare direttamente la tabella degli ordini condivisa. Se il destinatario necessita di una copia locale per analisi o carichi di lavoro a valle, può materializzare incrementalmente i dati in una Delta table. La configurazione avviene una sola volta ed è suddivisa tra fornitore e destinatario.
Crea la tabella di origine con Change Data Feed attivo, quindi pubblicala in una condivisione WITH HISTORY. La cronologia è ciò che rende la tabella un'origine incrementale. Senza di essa, il destinatario ottiene solo un'istantanea statica. L'abilitazione di Change Data Feed (delta.enableChangeDataFeed) è facoltativa per la lettura incrementale principale descritta di seguito. Attivala solo se prevedi di utilizzare in seguito il percorso per upsert ed eliminazioni. Se utilizzi Databricks Runtime 19 o versioni successive, puoi saltare la proprietà e affidarti a Change Data Feed automatico, che calcola le modifiche a livello di riga al momento della query per le tabelle con il tracciamento delle righe abilitato e funziona con Databricks-to-Databricks OpenSharing.
Registra il destinatario tramite il suo identificatore di condivisione, quindi concedi la condivisione:
Si prega di notare che CREATE SHARE e CREATE RECIPIENT richiedono privilegi a livello di metastore, non solo i privilegi di amministratore del workspace.
La condivisione ora viene visualizzata come un fornitore nel tuo metastore. Montala come catalogo e potrai leggerla come qualsiasi altra tabella.
Con la condivisione montata come shared_in, la lettura è gestita da Structured Streaming.
Trigger.AvailableNow elabora tutti i dati attualmente disponibili e non elaborati utilizzando uno o più micro-batch e quindi termina. Questo lo rende ideale per i processi incrementali pianificati. Invece di una pianificazione fissa, puoi anche configurare il processo con un trigger di aggiornamento della tabella sulla tabella condivisa, in modo che il processo venga eseguito solo quando il fornitore esegue il commit di nuovi dati. Si consiglia di archiviare il checkpoint in uno storage cloud durevole come UC Volumes.
Nota: questo esempio legge una tabella condivisa tramite Delta Sharing, quindi utilizza ignoreChanges. Se esegui lo streaming direttamente da una Delta table regolare, i runtime Databricks più recenti utilizzano skipChangeCommits per l'elaborazione basata solo sull'aggiunta.
Per dimostrare che è effettivamente incrementale, esegui lo stream una volta rispetto a un set iniziale di righe. Eseguilo di nuovo dopo aver aggiunto nuove righe sul fornitore. La cronologia dei commit della tabella di destinazione mostrerà due scritture in streaming:
Version | Operation | Operation Metrics |
|---|---|---|
2 | STREAMING UPDATE | numRemovedFiles: 0, numOutputRows: 200, numOutputBytes: 2307, numAddedFiles: 1 |
1 | STREAMING UPDATE | numRemovedFiles: 0, numOutputRows: 1000, numOutputBytes: 4958, numAddedFiles: 1 |
0 | CREATE TABLE | {} |
La seconda esecuzione ha scritto 200 righe, non 1200. Il checkpoint ha tenuto traccia dell'ultima versione letta, quindi l'esecuzione ha estratto solo quelle nuove.
ignoreChanges è adatto per questo pattern di arrivo incrementale, ma potrebbe riemettere righe e non propaga le eliminazioni. Se i consumer a valle necessitano di inserimenti, aggiornamenti ed eliminazioni, leggi invece il Change Data Feed della tabella condivisa:
Quando viene aggiunto un nuovo cliente, il fornitore non deve creare un nuovo endpoint SFTP o implementare un nuovo processo di esportazione. Il fornitore concede al destinatario l'accesso alla condivisione. Il destinatario utilizza lo stesso schema generale mantenendo separate le proprie tabelle di destinazione e la business logic a valle. La differenza operativa è significativa:
Tutto quanto sopra funziona perché il destinatario esegue Spark su Databricks, che è ciò che sblocca lo streaming con checkpoint e CDF. Altri destinatari (pandas, strumenti di BI, Snowflake) si connettono tramite OpenSharing con un file di credenziali anziché con un ID di condivisione del metastore.
Se eseguono OSS Spark, la lettura è lo stesso codice spark.readStream tramite il connettore delta-sharing. Senza Spark, non c'è uno stream continuo, quindi estraggono le modifiche per intervallo di versioni:
Un primo progetto utile non deve risolvere ogni scenario di condivisione. Inizia con una Delta table regolare e un destinatario.
Prima di implementare lo stream, rispondi a queste domande:
Una volta chiarite queste risposte, valida il flusso end-to-end più piccolo possibile: crea la condivisione, leggi un'istantanea iniziale, scrivi una nuova riga, aggiorna una riga, elimina una riga e conferma che il destinatario veda i risultati attesi.
Questo è il cambiamento pratico reso possibile da OpenSharing. Invece di creare un nuovo flusso di lavoro di trasporto e inserimento per ogni cliente, puoi stabilire un contratto di condivisione dei dati governato e consentire a ciascun destinatario di consumare i dati attraverso il percorso adatto alla sua piattaforma.
Nuovo su OpenSharing? Inizia con Cos'è OpenSharing? e Crea condivisioni per OpenSharing, quindi configura il flusso end-to-end minimo descritto sopra per una tabella e un destinatario.
(Questo post sul blog è stato tradotto utilizzando strumenti basati sull'intelligenza artificiale) Post originale
Iscriviti al nostro blog e ricevi gli ultimi articoli direttamente nella tua casella di posta.