Eclipse Aerospace semplifica l'acquisizione dei dati di volo con Databricks

Man mano che le organizzazioni nei settori aerospaziale, manifatturiero, delle scienze della vita e in altri settori ad alta intensità di dati modernizzano le loro operazioni, portano una maggiore quantità di dati provenienti da sensori, telemetria e dati scientifici nelle piattaforme di dati. Tuttavia, gran parte di questi dati rimane difficile da utilizzare su scala perché risiede in formati specializzati o proprietari che i motori di analisi tradizionali non possono leggere direttamente. Questo crea una sfida pratica per il settore: i team devono scegliere tra la creazione di trasformazioni personalizzate lente e ad alto consumo di memoria, la gestione di connettori specializzati o l'aggiunta di fasi di decodifica extra prima che i dati possano essere analizzati.

Le origini dati personalizzate di PySpark offrono una via d'uscita più semplice. Consentono ai team di importare formati specializzati in Spark utilizzando Python, senza dover creare un connettore separato da zero. Per Eclipse Aerospace, un produttore e operatore di jet leggerissimi, la collaborazione con Databricks ha offerto un modo per rendere quasi due decenni di dati di volo proprietari più accessibili ai centri di assistenza e agli ingegneri senza aggiungere un'infrastruttura costosa e sempre attiva. Databricks ha fornito la piattaforma di dati serverless e l'integrazione con Spark, mentre Eclipse Aerospace ha messo a disposizione i suoi decodificatori di dati di volo esistenti e le sue competenze aerospaziali.

Insieme, Eclipse Aerospace e Databricks hanno sviluppato un percorso più diretto dai dati di volo all'analisi, riducendo la latenza, la duplicazione e l'overhead operativo del precedente processo in più fasi, pur preservando la flessibilità di Databricks Serverless.

La sfida dell'ingestione dei dati per Eclipse Aerospace

Eclipse Aerospace si affida a Databricks per archiviare quasi 20 anni di dati di volo, per un totale di oltre 3TB. Supporta i suoi centri di assistenza e i suoi ingegneri con dati provenienti da oltre un milione di ore di volo, reperiti e verificati da più fonti, tra cui: telemetria satellitare quasi in tempo reale; grandi caricamenti massivi di intere cronologie di voli; e origini dati esterne. Tutta la telemetria è archiviata in formati di file proprietari che richiedono decodificatori personalizzati, già sviluppati in più linguaggi, tra cui Rust e Scala.

Prima che le origini dati personalizzate di PySpark fossero disponibili in Databricks Runtime 15.4 LTS, il team di Eclipse Aerospace aveva tre opzioni praticabili per decodificare e caricare il proprio tipo di file proprietario in Databricks:

1. Caricare i file come tipi binari e dividerli utilizzando funzioni definite dall'utente (UDF) Python
2. Usare un decodificatore personalizzato in Scala su un'istanza di calcolo dedicata
3. Decodificare i file in un tipo di file intermedio prima di caricarli in Databricks

Per i tipi di file personalizzati semplici, le UDF sono un'opzione naturale. Scambiano le prestazioni con la leggibilità. Tuttavia, laddove il formato del messaggio ha una lunghezza variabile, le UDF pongono un rischio reale di esaurimento della memoria sui messaggi lunghi e non potevano essere utilizzate. Il decodificatore Scala esistente poteva essere utilizzato per implementare un Data Source v1/v2 nativo, ma il suo utilizzo in Databricks richiedeva un'istanza di calcolo dedicata. Gli utenti si aspettavano di poter caricare e visualizzare i dati rapidamente, il che avrebbe richiesto di mantenere questa istanza di calcolo sempre attiva, rappresentando un costo maggiore rispetto alle altre opzioni serverless. Questo ha lasciato al team l'ultima opzione di decodificare i file in un formato di file intermedio, in questo caso Avro, tramite un decodificatore basato su Rust su AWS Lambda.

Questa soluzione era economica, ma:

• Ha introdotto latenza dal modello di ingestione a hop multipli
• Ha duplicato la logica dello schema tra il decodificatore Lambda e i lavori Spark a valle che consumavano l'output Avro
• È costata al team l'osservabilità della pipeline; l'UI di Spark e le metriche dei lavori vedevano solo il mondo a partire dalla zona di atterraggio di Avro, quindi gli ingegneri che eseguivano il debug di un'ingestione lenta o fallita dovevano andare a cercare i log di Lambda in una parte completamente diversa della piattaforma
• Ha comportato il pagamento per l'archiviazione sia dei file binari originali sia delle loro copie Avro

Perché Databricks e le origini dati personalizzate di PySpark sono importanti

Databricks era già la piattaforma di Eclipse Aerospace per l'archiviazione e l'elaborazione dei dati di volo. Il vantaggio decisivo è stato la combinazione dell'esecuzione nativa di Spark, dell'origine dati personalizzata di PySpark e di Databricks Serverless. Databricks Serverless alloca e gestisce automaticamente le risorse di calcolo on-demand, quindi Eclipse Aerospace non ha avuto bisogno di effettuare il provisioning o mantenere in esecuzione un'infrastruttura dedicata. L'origine dati personalizzata di PySpark ha reso possibile racchiudere il decodificatore Rust esistente senza costruire un connettore JVM. Poiché la decodifica ora viene eseguita all'interno di Spark, la stessa piattaforma fornisce l'UI di Spark, le metriche dei lavori e i flussi di lavoro Delta a valle, invece di affidare la fase critica di ingestione a una pipeline Lambda separata.

La maggior parte dell'ingegneria dei dati si basa su formati di dati comuni, ma alcuni dei dati più preziosi rimangono nascosti in formati a cui PySpark non può accedere direttamente, come HDF5 per i dati scientifici e standard industriali come ARINC 429 per i dati aerospaziali. Databricks Serverless offre ai team un modo flessibile per sbloccare tali dati su scala, mentre le implementazioni basate su Rust possono offrire prestazioni superiori a un approccio puramente Python.

Le origini dati personalizzate di PySpark consentono di definire un formato di file leggibile/scrivibile in puro Python, utilizzabile come qualsiasi altra origine dati. Il suo vantaggio principale è la semplicità, che consente agli utenti di definire funzioni per la lettura e la scrittura in puro Python, oltre a definire caratteristiche importanti come la parallelizzazione. Il team di Eclipse Aerospace ha fatto un ulteriore passo avanti: creando un pacchetto Python basato su Rust, ha offerto il doppio delle prestazioni registrate nei benchmark rispetto a un'origine dati puramente Python basata su PyArrow. Rispetto a un'implementazione Python nativa senza Arrow, ha offerto quasi 20 volte le prestazioni.

Il seguente repository di codice di esempio illustra un'origine dati personalizzata per ARINC 429. Il codice di accompagnamento include decisioni di progettazione che vanno oltre lo scopo di questo articolo, insieme a test di benchmark che è possibile eseguire autonomamente.

Come è diventato risolvibile con le origini dati personalizzate di PySpark

Apache Spark 4.0 ha introdotto un'API Python Data Source (citata nella documentazione di Databricks come origini dati personalizzate di PySpark) che consente di registrare un formato personalizzato per dati batch o in streaming come:

Introdotta in Apache Spark 4.0 e disponibile su Databricks, l'API è generalmente disponibile a partire da Databricks Runtime 15.4 LTS ed è supportata anche nell'ambiente Serverless 2 e successivi. Poiché l'API è implementata in Python, non richiede un connettore JVM o un JAR separato.

L'API è un'interfaccia semplificata per la creazione di origini dati native, in cui l'utente implementa alcune classi e le registra con:

Questi decodificatori sono disponibili in due varianti: Python nativo e formato Apache Arrow, quest'ultimo la scelta di gran lunga più performante. Apache Arrow è un "formato di memoria colonnale indipendente dal linguaggio per dati piatti e nidificati", il che lo rende un'interfaccia ideale tra diversi linguaggi che condividono oggetti di dati. L'API ne trae vantaggio consentendo agli utenti di restituire direttamente oggetti pyarrow.RecordBatch, e Spark li consumerà come batch anziché materializzare un oggetto InternalRow per riga.

Questo ha sollevato una domanda: Python potrebbe fungere da livello di controllo, con Rust che si occupa del lavoro pesante della decodifica mantenendo il formato Apache Arrow per tutto il tempo?

Panoramica dell'architettura della soluzione

L'architettura alla base di questo è un singolo connettore logico con un percorso di decodifica costituito quasi interamente da codice compilato e un livello Python abbastanza sottile da apportare modifiche minime all'SDK Rust esistente di Eclipse Aerospace. Questo sottile livello Python è costruito con PyO3/Maturin come pacchetto autonomo, importato nelle pipeline e utilizzato proprio come qualsiasi altra origine. Per garantire le migliori prestazioni e la separazione delle responsabilità, il team ha seguito alcuni principi architetturali.

Il decodificatore non sa che Spark esiste. Il crate Rust principale non ha dipendenze da PyO3, Arrow o Spark e accetta solo byte, restituendo record strutturati. Ciò ha permesso a Eclipse Aerospace di trasferire la logica di decodifica esistente quasi invariata, poiché l'SDK non ha dovuto essere riscritto per questo, ma solo racchiuso.

Arrow è l'unica cosa che attraversa il confine dell'interfaccia di funzione straniera (FFI) e lo attraversa senza copie. Il livello di raccordo in Rust crea un RecordBatch Arrow utilizzando arrow-rs, quindi lo consegna a Python come pyarrow.RecordBatch tramite l'interfaccia dati C di Arrow. In questo modo, non c'è alcuna fase di serializzazione, poiché nessun byte viene ricodificato e nessun oggetto Python per riga viene creato. I buffer allocati da Rust sono i buffer letti da Python.

Il livello PySpark è volutamente leggero. Analizza le opzioni, decide come partizionare (uno InputPartition per input nell'esempio di riferimento) e definisce lo schema. Nei data source personalizzati di PySpark, DataSource, DataSourceReader e i loro equivalenti in streaming devono essere tutti serializzabili, in modo che Spark possa eseguire il pickle dell'istanza del reader e inviarla a un worker. Rendendo leggero il livello PySpark, l'istanza DataSourceReader che Spark inserisce nel pickle e invia a un worker contiene sempre e solo lo stato Python normale (percorsi di file, un intero per la dimensione del batch). L'oggetto Rust basato su PyO3 viene costruito all'interno di read(), dopo che il reader è già arrivato sul worker. Non fa mai parte di ciò che viene serializzato con il pickle.

Il codice nativo viene distribuito come wheel Python, compilata con Maturin. Su Databricks, tale wheel è un artefatto di compilazione, associato a un ambiente di job o notebook come normale dipendenza Python. Poiché l'implementazione è compilata, la compatibilità di versione e architettura richiede attenzione, sebbene un processo CI/CD adeguato possa gestire questo aspetto come dimostrato nel repository di esempio.

Come viene eseguito il connettore in Databricks

Su Databricks Serverless, il percorso di runtime è semplice:

1. Un file di telemetria proprietario viene salvato nello storage cloud.
2. Una lettura Spark su Databricks richiama il data source personalizzato PySpark registrato per il formato personalizzato.
3. Spark in esecuzione su Databricks crea un reader serializzabile contenente il percorso di input e le opzioni di lettura, quindi pianifica le partizioni di input sui worker.
4. Ogni worker costruisce il decoder Rust all'interno di read() e passa i byte del file al decoder compilato.
5. Rust restituisce Arrow RecordBatch tramite Arrow C Data Interface, evitando la serializzazione e gli oggetti Python per riga.
6. Spark in esecuzione su Databricks utilizza i batch e scrive i dati decodificati nelle tabelle downstream come Delta.
7. Gli errori di decodifica, i record non validi, le fasi della UI di Spark e le metriche dei job rimangono visibili all'interno del flusso di lavoro di Databricks.

Questa è la differenza chiave rispetto alla progettazione precedente: il file originale viene decodificato come parte della lettura Spark anziché essere copiato prima in una zona di staging Avro intermedia.

Qual è stato l'impatto per Eclipse Aerospace

Sostituendo il flusso di lavoro Lambda e Avro a più fasi con una singola lettura nativa Spark, Eclipse Aerospace ha reso i dati di volo disponibili più rapidamente riducendo al contempo l'infrastruttura, lo storage e i costi operativi. Il nuovo connettore di data source personalizzati PySpark nativo per Arrow porta inoltre la decodifica all'interno di Spark, migliorando la visibilità sulle prestazioni della pipeline e consentendo di gestire la logica dello schema e della qualità dei dati in un'unica posizione.

L'implementazione del data source personalizzato ha notevolmente semplificato l'infrastruttura di Eclipse Aerospace e gli effetti si manifestano in quattro aree distinte: latenza, costo, osservabilità e tempo di sviluppo.

Latenza. La compressione di una pipeline a più hop (caricamento, Lambda, decodifica, scrittura Avro, lettura Avro, caricamento in Delta) in un'unica lettura nativa Spark elimina diversi hop di rete e diversi punti in cui un file rimane in coda in attesa che la fase successiva lo prelevi. Il nuovo connettore è in grado di elaborare i record entro un minuto dal caricamento del file, incluso il tempo di avvio, ed è solitamente in grado di elaborare batch grandi di diversi gigabyte in parallelo in meno di due minuti grazie al parallelismo.

Costo. La rimozione dell'hop Lambda e dell'intermediario Avro ha eliminato un intero livello di infrastruttura che Eclipse Aerospace doveva eseguire, monitorare e pagare. Non devono più pagare per memorizzare due copie di ogni volo, il file binario originale e la sua conversione Avro, che su quasi due decenni di storico dei voli conservato rappresenta una quantità significativa di storage duplicato. Ciò significa che la maggior parte dei file rimane nello storage Glacier su AWS, con decodifiche facilmente ripetibili se i dati originali dovessero servire di nuovo.

Osservabilità. Poiché la fase di decodifica avviene ora all'interno di Spark stesso, viene visualizzata nella UI di Spark e nelle metriche di job standard come qualsiasi altra fase di un job. Eclipse Aerospace utilizza il parametro personalizzato "badRecordsPath", che imita il loader CSV/JSON, per acquisire i dati formattati in modo errato e mantiene ampie pipeline di qualità dei dati che acquisiscono statistiche direttamente dal decoder.

Tempo di sviluppo. La logica dello schema veniva definita due volte: una volta nel decoder Lambda che produceva Avro e di nuovo nei job Spark downstream che lo utilizzavano. Qualsiasi modifica al formato del messaggio significava aggiornarli entrambi, in due codebase diverse, con il rischio che le due definizioni divergessero. Con un connettore singolo e con versioni gestite, lo schema viene definito una volta, accanto alla logica di decodifica stessa. Ciò significa anche che lo stesso pattern architetturale, un sottile DataSource Python che incapsula un decoder Rust, è ora riutilizzabile per gli altri formati proprietari supportati da Eclipse Aerospace, anziché essere qualcosa da reinventare per ogni formato.

L'esperienza di Eclipse Aerospace dimostra come sia possibile integrare formati di dati specializzati in Databricks senza creare una piattaforma di inserimento dati separata attorno ad essi. Mantenendo il decoder in Rust, utilizzando Python come livello di integrazione e passando i record tramite Apache Arrow, il data source personalizzato legge i file originali direttamente in Spark e rimane compatibile con la flessibilità di Databricks Serverless.

Per i team che lavorano con formati proprietari o specifici del settore, il pattern riutilizzabile è semplice. Gli sviluppatori possono mantenere la logica di decodifica specifica del dominio vicina al formato di origine, esporla tramite un data source personalizzato PySpark e lasciare che Spark gestisca il partizionamento, l'esecuzione, l'osservabilità e l'elaborazione downstream.

Vuoi iniziare? Dai un'occhiata alle risorse seguenti:

• PySpark su Databricks
• Data source personalizzati PySpark
• Connettiti al compute serverless

Domande frequenti

Che cos'è un PySpark Custom Data Source?

Un data source personalizzato PySpark è un data source Spark definito dall'utente e implementato in Python. Consente ai team di leggere formati di file specializzati come parte di un flusso di lavoro Spark standard senza dover creare un connettore JVM separato.

Quale problema di inserimento dati risolve un connettore di PySpark Custom Data Source nativo per Arrow?

Il connettore di data source personalizzato PySpark nativo per Arrow acquisisce la telemetria di volo proprietaria in Databricks senza fare affidamento su una pipeline a più fasi, su compute dedicato sempre attivo o su copie Avro duplicate dei file originali.

Che cos'è Databricks Serverless?

Databricks Serverless è un servizio di compute completamente gestito che consente di eseguire carichi di lavoro senza dover effettuare il provisioning o gestire alcuna infrastruttura nel proprio account cloud. Databricks assegna, scala e ottimizza automaticamente le risorse di compute per te.

In che modo Databricks Serverless si connette al connettore di PySpark Custom Data Source nativo per Arrow?

Il connettore utilizza i data source personalizzati Pyspark per eseguire la logica di lettura dei file all'interno di Spark su compute serverless. Python gestisce il flusso di controllo, mentre Rust e Apache Arrow gestiscono la decodifica ad alte prestazioni e il trasferimento dei dati. Ciò consente di leggere i file proprietari direttamente senza compute dedicato sempre attivo o una pipeline Avro intermedia. I data source personalizzati PySpark sono disponibili in Databricks Runtime 15.4 LTS e versioni successive, oppure nell'ambiente serverless versione 2.

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