Ecco Apache Spark® 4.3 - Ora disponibile in Databricks Runtime 19

Introduzione

Apache Spark 4.3 introduce l'elaborazione con stato nella Modalità in tempo reale, ottenendo una latenza P99 end-to-end inferiore a 150 ms su tre benchmark con stato rappresentativi. Le aggregazioni in streaming, la deduplica e l'elaborazione con stato personalizzata possono ora essere eseguite nella modalità di esecuzione a bassa latenza di Structured Streaming, estendendola alle applicazioni che devono memorizzare il contesto tra i vari eventi.

L'impatto va oltre la velocità delle pipeline. Per i carichi di lavoro supportati, non è più necessario un motore di streaming separato solo per ottenere una latenza dell'ordine dei millisecondi. Dal rilevamento delle frodi all'attività degli utenti aggiornata continuamente e al monitoraggio dei dispositivi, gli sviluppatori possono creare applicazioni con stato utilizzando le familiari API di Structured Streaming nello stesso motore aperto che usano per l'elaborazione batch, l'analisi e l'AI.

Lo streaming in tempo reale con stato è la novità principale di Spark 4.3, ma i miglioramenti interessano tutto il motore. La release offre cinque vantaggi chiave:

• Crea applicazioni in tempo reale con stato: utilizza l'aggregazione in streaming, la deduplica e transformWithState nella Modalità in tempo reale per mantenere il contesto man mano che arrivano gli eventi.
• Conserva la cronologia con SCD Type 2 Auto CDC: monitora come cambiano i record nel tempo utilizzando Python, SQL e Spark Connect, con una minore logica di elaborazione delle modifiche personalizzata
• Ottieni iterazioni più rapide sui dati di piccole dimensioni: Project Feather punta a ridurre l'overhead di esecuzione non necessario, mentre la memorizzazione nella cache nativa di Arrow supporta un riutilizzo dei dati più efficiente.
• Collega più applicazioni a Spark: i miglioramenti all'osservabilità di Spark Connect, all'integrazione JDBC e a Python rendono l'esecuzione remota più semplice da sviluppare e gestire.
• Esprimi analisi più ricche direttamente in SQL: le trasformazioni VARIANT, JSON_TABLE, i join temporali e un'anteprima dei timestamp in nanosecondi estendono l'elaborazione nativa dei dati nidificati e orientati al tempo.

Insieme, queste modifiche aiutano gli sviluppatori a utilizzare un unico motore aperto per esplorare i dati, mantenere il contesto variabile e alimentare applicazioni analitiche e di AI.

Spark 4.3 segna anche l'inizio di una cadenza trimestrale di rilascio delle funzionalità, passando da una release ogni sei mesi a una ogni tre mesi. Questo ciclo più breve consente agli utenti di usufruire prima dei miglioramenti completati e offre alla community opportunità più frequenti per fornire feedback e definire i prossimi passi.

Streaming in tempo reale: applicazioni con stato e latenza su scala millisecondi

Dall'elaborazione degli eventi al mantenimento del contesto

Un segnale di frode non dipende solo dall'ultima transazione, ma anche dall'attività che l'ha preceduta. La personalizzazione deve riflettere il comportamento recente di un utente. Il monitoraggio dei dispositivi deve aggiornare contatori e aggregati man mano che arrivano nuove letture. Queste applicazioni richiedono sia bassa latenza che stato.

Spark 4.3 estende la Modalità in tempo reale a questi pattern di elaborazione con stato, rendendolo un motore di streaming completo. Aggiunge il supporto per l'aggregazione in streaming, dropDuplicates e Scala/Java transformWithState (SPARK-58635, SPARK-57228). Gli sviluppatori possono mantenere lo stato basato su chiave e aggiornare i risultati utilizzando il modello di programmazione esistente di Structured Streaming, anziché adottare un altro motore esclusivamente per ottenere una latenza inferiore.

Latenza P99 inferiore a 150 ms su tre benchmark con stato

Nei benchmark riportati in SPARK-58884, tutti e tre i carichi di lavoro con stato rappresentativi hanno raggiunto una latenza end-to-end P99 inferiore a 150 millisecondi:

Questi risultati dimostrano un'esecuzione a bassa latenza tra deduplica, aggregazione a finestra ed elaborazione con stato personalizzata. Si tratta di misurazioni dei tre carichi di lavoro testati, non di una garanzia di latenza per ogni pipeline.

API familiari, esecuzione a bassa latenza

La Modalità in tempo reale si basa su Structured Streaming anziché introdurre un modello di programmazione separato. Gli operatori con stato continuano a utilizzare l'archivio di stato e i meccanismi di checkpoint esistenti di Spark per garantire durabilità e ripristino.

Esempio: mantieni un conteggio progressivo per ciascun dispositivo. Dato un DataFrame in streaming basato su Kafka events con una colonna device_id:

Se arrivano eventi per i dispositivi A, A e B, Spark mantiene lo stato e aggiorna i conteggi a A=2 e B=1. Quando arriva un altro evento per A, il suo stato viene aggiornato a A=3.

Un motore di esecuzione creato per il flusso di dati continuo

Il supporto della Modalità in tempo reale con stato ha richiesto un cambiamento fondamentale nel modo in cui Spark pianifica e collega le fasi. Con uno shuffle di blocco tradizionale, una fase di produzione deve terminare prima che la fase downstream possa iniziare. Quel modello non è in grado di supportare un flusso di dati continuo tra attività di streaming a esecuzione prolungata.

Spark 4.3 introduce la pianificazione nativa delle fasi concorrenti e gli shuffle pipelined, consentendo alle fasi produttrici e consumatrici di essere eseguite contemporaneamente (SPARK-54699, SPARK-58076, SPARK-56664). Streaming Shuffle implementa questa funzionalità per la Modalità in tempo reale, trasferendo i record in modo incrementale all'attività downstream responsabile di ciascuna chiave invece di attendere il completamento dell'intera fase di produzione.

Concettualmente, l'esecuzione passa da:

Fase del produttore → materializza shuffle → fase del consumatore

a:

Fase del produttore ⇄ streaming shuffle ⇄ fase del consumatore

Spark 4.3 aggiunge PipelinedShuffleDependency, un nuovo tipo di dipendenza di shuffle che consente ai consumatori di iniziare la lettura mentre i produttori sono ancora in esecuzione (SPARK-58185). Lo scheduler tratta le fasi collegate da queste dipendenze come un gruppo pipelined, un'unità di pianificazione coordinata. Le sue fasi vengono pianificate insieme e un errore in una fase causa il fallimento dell'intero gruppo (SPARK-58263, SPARK-58398). In breve, vengono eseguite insieme e falliscono insieme.

Poiché tutte le fasi in un gruppo pipelined vengono eseguite contemporaneamente, Spark ammette il gruppo solo quando sono disponibili slot di attività sufficienti per tutte le sue attività. In caso contrario, fallisce rapidamente anziché bloccarsi.

Spark instrada ogni shuffle a un gestore di shuffle in base al suo tipo di dipendenza:

• Un normale ShuffleDependency utilizza un gestore di shuffle di blocco, che materializza tutto l'output dello shuffle prima che le fasi downstream lo leggano. SortShuffleManager è l'implementazione incorporata.
• Un PipelinedShuffleDependency utilizza un gestore di shuffle pipeline, che rende l'output disponibile man mano che viene prodotto. StreamingShuffleManager è l'implementazione integrata (SPARK-57141).

Il tipo di dipendenza definisce il contratto di esecuzione; il gestore di shuffle fornisce il meccanismo di trasferimento dei dati. Lo scheduler coordina le fasi in pipeline senza dover sapere quale carico di lavoro le ha create.

Questa base va oltre la modalità in tempo reale. La pianificazione concorrente delle fasi, lo scambio di dati in pipeline e l'esecuzione coordinata sono funzionalità generiche del motore Spark, non un'infrastruttura di streaming isolata. In Spark 4.3, consentono l'elaborazione stateful a bassa latenza. A lungo termine, forniscono una base per altri carichi di lavoro che devono andare oltre il tradizionale modello di esecuzione fase per fase.

Per informazioni di background sul design della modalità in tempo reale e sugli scenari di applicazione, consulta l' introduzione alla modalità in tempo reale. Quel post descrive l'implementazione di Databricks; l'ambito delle funzionalità di Spark 4.3 e i risultati dei benchmark qui discussi sono quelli elencati sopra.

Project Feather: Iterazione più rapida su piccoli dati

Lo sviluppo assistito da LLM segue spesso un ciclo di feedback breve: genera una trasformazione, eseguila su un campione, ispeziona i risultati e perfeziona il codice. Ciascuna iterazione può elaborare solo un file piccolo o poche righe, eppure la pianificazione delle attività e lo shuffle possono costare più del calcolo stesso.

Project Feather si concentra su questo overhead di esecuzione fisso, rendendo Spark più reattivo per l'esplorazione e la validazione del codice SQL e DataFrame su piccoli dataset. Gli sviluppatori possono utilizzare le stesse API man mano che passano da un campione a carichi di lavoro più ampi, sia che scrivano il codice autonomamente sia che lavorino con un assistente AI.

Spark 4.3 introduce l'esecuzione a singolo task senza shuffle e opzionale per i frammenti di query idonei, di piccole dimensioni e a partizione singola (SPARK-57851). Quando una scansione idonea segnala che il suo output si trova già in una singola partizione, le operazioni downstream supportate possono evitare uno shuffle non necessario.

Esempio: esegui e ispeziona una piccola aggregazione. Utilizza un piccolo dataset Parquet memorizzato come file singolo, con una colonna di categoria:

Sei casi:

category, Exchange, SPARK_LOCAL_CONNECT_REUSE=1, zip, counter_diff and jaro_winkler_similarity.

Il risultato è un conteggio per ciascuna categoria. Per una query idonea, il piano fisico può omettere l'`Exchange` che altrimenti introdurrebbe uno shuffle. Questo supporta lo stesso flusso di lavoro di esecuzione e ispezione, sia che la trasformazione sia stata scritta da uno sviluppatore sia che sia stata suggerita da un assistente di codifica.

Con i limiti di idoneità predefiniti, le scansioni di file idonee leggono un piccolo file non vuoto che rientra in una singola partizione, mentre le relazioni locali in-memory idonee contengono fino a 1.000 righe. Le forme di query supportate includono scansioni selezionate, aggregazioni, operazioni distinct, finestre, ordinamenti e limiti.

Feather aggiunge inoltre la memorizzazione nella cache opzionale dei Dataset nativa di Arrow (SPARK-57268), salvando i batch in cache in formato Arrow con compressione e statistiche sulle colonne. Gli input basati su Arrow possono evitare conversioni non necessarie, supportando flussi di lavoro che esplorano ripetutamente gli stessi dati. Sia l'ottimizzazione del task singolo che la cache di Arrow sono opzionali. Per attivarle, imposta la configurazione statica spark.sql.cache.serializer=org.apache.spark.sql.execution.columnar.ArrowCachedBatchSerializer all'avvio di una nuova applicazione Spark. Una ripartizione locale in-process più ampia è un lavoro successivo previsto per la prossima release (SPARK-57399).

Questi miglioramenti sono solo l'inizio. La community di Spark continuerà a investire sulle prestazioni con dati di piccole dimensioni nelle future release, riducendo l'overhead di esecuzione e accorciando i cicli di feedback per l'esplorazione interattiva e lo sviluppo assistito da LLM.

SCD Type 2 in Open-Source Auto CDC: Mantieni la cronologia man mano che i dati cambiano

Una tabella dello stato corrente può dirti dove vive un cliente oggi, ma non a quale regione apparteneva quando è stato effettuato un ordine. Reportistica, auditing e analisi storiche spesso richiedono entrambi. Spark 4.3 aggiunge il supporto end-to-end per il Slowly Changing Dimension (SCD) Type 2 ad Auto CDC (SPARK-56249), in modo che le pipeline possano tracciare la cronologia di ogni record anziché sovrascrivere l'ultimo valore a ogni modifica. La funzionalità è disponibile tramite Python, SQL e Spark Connect (SPARK-58321, SPARK-58247, SPARK-58261) e consente agli utenti di applicare automaticamente le modifiche anziché eseguire operazioni di merge manuali.

Esempio: traccia lo stato e la regione del cliente nel tempo. Data una sorgente di pipeline customer_changes con ID cliente, ora dell'evento, stato e regione:

In alternativa, questo può essere espresso in SQL:

In una tabella di output SCD-2, ogni chiave univoca è rappresentata da una sequenza di record, ciascuno dei quali contrassegna l'inizio e la fine della sua validità utilizzando due colonne di metadati aggiuntive, __START_AT e __END_AT. Ad esempio, quando un record di modifica indica che il cliente con id 17 si è spostato dalla regione West alla East, la pipeline conserva il record con la regione precedente West e contrassegna la sua fine come l'ora dell'evento del record di modifica; inserisce un nuovo record con la regione East e l'ora dell'evento del record come inizio. Un'analisi può quindi utilizzare la regione applicata nel punto rilevante della sequenza temporale degli eventi.

Dietro le quinte, Auto CDC collega la definizione del flusso dichiarativo alla derivazione dello schema di destinazione e all'elaborazione di stream dedicata per SCD Type 2 (SPARK-58319, SPARK-57395, SPARK-58321). Le chiavi identificano l'entità, la sequenza determina l'ordine delle modifiche e le colonne di tracciamento della cronologia specificano quali modifiche agli attributi creano versioni storiche. Il motore convalida anche le colonne riservate, i tipi di sequenziamento e la configurazione di tracciamento della cronologia (SPARK-57251, SPARK-58313, SPARK-58391) e riconcilia micro-batch più ristretti con uno schema di destinazione evoluto (SPARK-58418).

Auto CDC applica ciascun micro-batch alla destinazione utilizzando MERGE, pertanto la tabella di destinazione deve trovarsi in un catalogo il cui connettore implementa le operazioni a livello di riga di Data Source V2. I formati di file integrati di Spark, incluso Parquet, non sono idonei. Pertanto, una scelta comune è Iceberg.

Spark Connect: più facile da sviluppare, monitorare e gestire

Le applicazioni e gli agenti AI devono invocare Spark dai propri processi, gestire le sessioni e collegare una richiesta non riuscita al lavoro in esecuzione sul server. Spark 4.3 rafforza Spark Connect per questi flussi di lavoro con la visibilità di sessioni e operazioni, l'attribuzione delle richieste e un supporto client più ampio.

Spark Connect invia le operazioni DataFrame come piani logici tramite gRPC, le esegue attraverso il motore Spark del server e trasmette in streaming i risultati in batch Arrow. Spark 4.3 aggiunge endpoint REST per elencare e ispezionare sessioni e operazioni di Connect, oltre a ID di operazione per tracciare l'esecuzione tra client e server (SPARK-57941, SPARK-58547). Per lo sviluppo locale PySpark su sistemi POSIX, l'opzione sperimentale `SPARK_LOCAL_CONNECT_REUSE=1` riutilizza un server Connect persistente tra le esecuzioni, evitando i costi generali di avvio ripetuti (SPARK-57787). Il wrapper di compatibilità SQLContext di PySpark funziona anche con Spark Connect (SPARK-57021).

Il driver JDBC di Spark Connect aggiunge il supporto per i risultati ARRAY, MAP e STRUCT e una gestione più ricca dei set di risultati (SPARK-57380, SPARK-57274), metadati di tipo SQL e un'ispezione dello schema più compatibile (SPARK-57627) e una gestione delle eccezioni migliorata (SPARK-57933, SPARK-58619). La copertura delle API si espande con DataFrame.zip, SCD Type 2 Auto CDC e tipi e letterali di timestamp in nanosecondi (SPARK-57247, SPARK-58247, SPARK-57160, SPARK-57161).

Separatamente, il lavoro della community include Spark Connect Gateway (SPARK-58455) per il routing consapevole della sessione e le operazioni multi-tenant, client Python basati su Rust e Rust (SPARK-58960) per un ulteriore percorso di implementazione del client e una proposta di individuazione delle capacità (SPARK-56398) per identificare le operazioni del server supportate prima di inviare una query. Questi sforzi hanno tempistiche di sviluppo separate e non sono funzionalità aggregate della versione 4.3.

PySpark: API più ricche e migliore esecuzione Python

I flussi di lavoro Python spesso costruiscono colonne derivate in passaggi separati, combinano i risultati e spostano i dati tra pandas, Arrow e Spark. Spark 4.3 rende gran parte di questo lavoro disponibile tramite le API DataFrame migliorando al contempo i percorsi di esecuzione sottostanti. DataFrame.zip aggiunge un modo per combinare proiezioni di colonne compatibili senza scendere alle API RDD, anche quando si utilizza Spark Connect (SPARK-55886, SPARK-57247).

Esempio: combina gli identificatori con un valore derivato. Crea due proiezioni dallo stesso input e unisci le loro colonne:

Il risultato associa id=1 a doubled=20 e id=2 a doubled=40. `zip` combina trasformazioni basate solo su proiezioni dello stesso piano sottostante; il filtraggio, l'ordinamento, i join o le aggregazioni al di sotto di tali proiezioni devono essere identici su entrambi i lati per preservare l'allineamento delle righe.

Il lavoro di esecuzione più ampio mira al trasferimento dei dati, alla creazione di batch e al calcolo nativo. I batch di input UDF basati su Pickle possono essere limitati dai byte anziché solo dal conteggio delle righe, offrendo un controllo migliore quando le righe variano in dimensioni (SPARK-57593). Le proiezioni batch pandas-on-Spark e ulteriori implementazioni native di NumPy riducono il lavoro evitabile (SPARK-58230, SPARK-58234), mentre la conversione da pandas a Spark aggiunge il supporto per le colonne PyArrow ChunkedArray (SPARK-46776).

La release aggiunge anche binding Python e Scala per le funzioni vettoriali esistenti (SPARK-57284), covarianza raggruppata e statistiche descrittive (SPARK-40336, SPARK-40337), operazioni di mediana e errore standard mobili e in espansione (SPARK-40341, SPARK-40340) e diagnostica show_versions (SPARK-40373). Le variabili locali dei traceback sono facoltative perché possono contenere dati sensibili (SPARK-57271).

Spark SQL: prepara un contesto migliore per l'analisi e l'IA

I payload degli eventi arrivano come documenti nidificati ed eventi ravvicinati possono richiedere una precisione superiore ai microsecondi. La preparazione di questi input non dovrebbe richiedere un livello di elaborazione personalizzato separato per ogni trasformazione. Spark 4.3 espande il linguaggio SQL nativo per dati semi-strutturati e analisi temporali, inclusi la manipolazione dei documenti, il confronto temporale e un'anteprima opzionale del timestamp in nanosecondi (SPARK-56822).

Esempio: distingui due eventi all'interno dello stesso microsecondo. Abilita l'anteprima (funzionalità non ancora stabile, potrebbero verificarsi modifiche sostanziali) e confronta i timestamp distanti un nanosecondo (SPARK-56969, SPARK-56965, SPARK-57211):

Il confronto restituisce true: la cifra finale viene preservata anziché scartata. L'anteprima supporta TIMESTAMP_NTZ(p) e TIMESTAMP_LTZ(p) a precisioni da 7 a 9.

Ciò ha richiesto molto più che accettare il nome di un nuovo tipo SQL. Il supporto ai nanosecondi si estende all'analisi, alle rappresentazioni dei valori interni, ai cast, alle espressioni, a Spark Connect, alla memorizzazione nella cache e all'integrazione delle origini dati (SPARK-56822). Le nuove funzioni timestamp_nanos e unix_nanos gestiscono le conversioni numeriche (SPARK-57526, SPARK-57527). Il supporto per l'archiviazione e i connettori rimane specifico per ogni percorso dati, quindi l'anteprima deve essere valutata end-to-end.

Trasforma payload JSON mutevoli in dati utilizzabili

Gli eventi applicativi, le esportazioni di documenti e le risposte dei modelli arrivano spesso come JSON nidificati. JSON_TABLE di Spark 4.3 (SPARK-58366) combina l'espansione degli array e l'estrazione di campi tipizzati, mentre le trasformazioni VARIANT consentono alle pipeline di aggiornare o rimuovere campi senza UDF di elaborazione documenti personalizzate (SPARK-57804, SPARK-57417). Insieme, semplificano la preparazione dei contenuti per le pipeline di analisi, arricchimento e recupero.

Esempio: estrai le sezioni di documenti pertinenti.

Questo produce una riga tipizzata che una pipeline può arricchire con metadati di origine prima di altre operazioni downstream. JSON_TABLE risolve i percorsi delle sue colonne attraverso un attraversamento condiviso di ogni elemento della riga, evitando un'analisi separata per ogni colonna. Separatamente, Spark estende il pushdown dell'estrazione di campi VARIANT alle query supportate che aggregano, ordinano o eseguono il join. (SPARK-58089).

Ulteriori attività su SQL/JSON (SPARK-58686) estendono questa direzione con JSON_VALUE, JSON_EXISTS e JSON_QUERY per l'estrazione scalare, i controlli di presenza e il recupero dei frammenti (SPARK-58685, SPARK-58687, SPARK-58723). La gestione esplicita dei valori mancanti e degli errori aiuterà le applicazioni a elaborare schemi mutevoli in modo più prevedibile. Si tratta di sviluppi successivi, non di ulteriori funzionalità della versione 4.3.

Associa gli eventi alle informazioni disponibili in quel momento

L'analisi storica e la valutazione dei modelli richiedono il contesto disponibile quando si è verificato un evento, non le informazioni aggiunte in seguito. SQL ASOF JOIN di Spark 4.3 esprime (funzione in anteprima, potrebbero verificarsi modifiche sostanziali) questa corrispondenza temporale direttamente, sostituendo i join di intervallo scritti manualmente seguiti dal ranking (SPARK-58092).

Esempio: ricostruisci le feature disponibili per una previsione. Qui, available_at registra quando una versione di una feature è diventata disponibile (SPARK-58122, SPARK-58123):

Una previsione alle 10:03 corrisponde alla versione della feature disponibile alle 10:00, non alle 10:05. Un percorso di esecuzione sort-merge dedicato rende questa corrispondenza un'operazione del motore anziché una logica di query specifica dell'applicazione. Il risultato è un arricchimento point-in-time più semplice per dati di addestramento, valutazioni e analisi degli incidenti, a condizione che i timestamp di disponibilità rappresentino accuratamente ciò che era noto all'epoca.

Miglioramenti notevoli e ringraziamenti

Spark 4.3 introduce anche nuove API e miglioramenti delle prestazioni per l'elaborazione dati quotidiana. SQL ottiene ANSI UNNEST per espandere gli array in righe, insieme alla funzione di finestra `counter_diff` per convertire contatori cumulativi in delta e `jaro_winkler_similarity` per i confronti di somiglianza tra stringhe. Le origini di file supportate aggiungono il supporto opzionale per la lettura di archivi tar, zip e 7z, eliminando una fase di estrazione separata. Il motore migliora l'esecuzione delle aggregazioni e abilita per impostazione predefinita la generazione di codice whole-stage per le operazioni UNION idonee. Un'ottimizzazione opzionale distribuisce le chiavi di join null tra le partizioni di shuffle nei join esterni e anti idonei. L'elaborazione di dati nidificati ottiene una maggiore rimozione dei campi non utilizzati all'interno di array di struct e la generazione di codice per un numero maggiore di funzioni di ordine superiore degli array. Le letture Parquet vettorializzate migliorano la conversione dei tipi e la decodifica dei dizionari.

Spark 4.3 riflette la forza della community di Apache Spark, con più di 200 contributori che hanno aiutato a risolvere oltre 1.100 ticket Jira. Ringraziamo tutti coloro che hanno contribuito con codice, revisioni, test, documentazione e feedback per rendere possibile questa release.

Inizia con Spark 4.3

Visita la pagina dei download di Apache Spark per i download della release e la documentazione, e consulta le guide di migrazione prima di eseguire l'aggiornamento. Per la disponibilità su Databricks, consulta le note di rilascio di Databricks Runtime.

(Questo post sul blog è stato tradotto utilizzando strumenti basati sull'intelligenza artificiale) Post originale