Eclipse Aerospace simplifie l'ingestion de données de vol avec Databricks

Alors que les entreprises de l'aérospatiale, de la fabrication, des sciences de la vie et d'autres secteurs intensifs en données modernisent leurs opérations, elles intègrent davantage de données de capteurs, de télémétrie et de données scientifiques dans leurs plateformes de données. Cependant, une grande partie de ces données reste difficile à utiliser à grande échelle, car elles résident dans des formats spécialisés ou propriétaires que les moteurs d'analyse grand public ne peuvent pas lire directement. Cela crée un défi industriel concret : les équipes doivent choisir entre créer des transformations personnalisées lentes et gourmandes en mémoire, maintenir des connecteurs spécialisés ou ajouter des étapes de décodage supplémentaires avant de pouvoir analyser les données.

Les sources de données personnalisées PySpark offrent une solution plus simple. Elles permettent aux équipes d'importer des formats spécialisés dans Spark à l'aide de Python, sans avoir à créer un connecteur séparé à partir de zéro. Pour Eclipse Aerospace, fabricant et opérateur de jets très légers, s'associer avec Databricks a permis de rendre près de deux décennies de données de vol propriétaires plus accessibles aux centres de service et aux ingénieurs, sans ajouter d'infrastructure coûteuse et toujours active. Databricks a fourni la plateforme de données serverless et l'intégration Spark, tandis qu'Eclipse Aerospace a apporté ses décodeurs de données de vol existants et son expertise aérospatiale.

Ensemble, Eclipse Aerospace et Databricks ont développé un chemin plus direct entre les données de vol et l'analytique, réduisant la latence, la duplication et la charge opérationnelle du processus multi-étape précédent tout en préservant la flexibilité de Databricks Serverless.

Le défi d'ingestion de données d'Eclipse Aerospace

Eclipse Aerospace s'appuie sur Databricks pour stocker près de 20 ans de données de vol, soit un total de plus de 3 To. L'entreprise prend en charge ses centres de service et ses ingénieurs avec des données issues de plus d'un million d'heures de vol, collectées et vérifiées à partir de plusieurs sources, notamment : la télémétrie satellite quasi en temps réel, les transferts massifs de historiques de vol complets, et des sources de données externes. Toute sa télémétrie est stockée dans des formats de fichiers propriétaires qui nécessitent des décodeurs personnalisés, déjà développés dans plusieurs langages, dont Rust et Scala.

Avant que les sources de données personnalisées PySpark ne soient disponibles dans Databricks Runtime 15.4 LTS, l'équipe d'Eclipse Aerospace disposait de trois options viables pour décoder et charger son type de fichier propriétaire dans Databricks :

1. Charger les fichiers en tant que types binaires et les diviser à l'aide de fonctions définies par l'utilisateur (UDF) Python
2. Utiliser un décodeur personnalisé en Scala sur un calcul dédié
3. Décoder les fichiers dans un type de fichier intermédiaire avant le chargement dans Databricks

Pour les types de fichiers personnalisés simples, les UDF sont une option naturelle. Elles sacrifient les performances au profit de la lisibilité. Cependant, lorsque le format du message a une longueur variable, les UDF présentent un risque réel de dépassement de mémoire sur les longs messages et n'ont pas pu être utilisées. Le décodeur Scala existant pouvait être utilisé pour implémenter une Data Source v1/v2 native, mais son utilisation dans Databricks nécessitait un calcul dédié. Les utilisateurs s'attendaient à pouvoir télécharger et afficher rapidement les données, ce qui aurait nécessité de maintenir ce calcul toujours actif, représentant un coût plus élevé que les autres options serverless. Cela a laissé à l'équipe la dernière option : décoder les fichiers dans un format de fichier intermédiaire, dans ce cas Avro, via un décodeur basé sur Rust sur AWS Lambda.

Cette solution était peu coûteuse, mais elle :

• A introduit de la latence en raison du modèle d'ingestion à plusieurs sauts
• A dupliqué la logique de schéma entre le décodeur Lambda et les travaux Spark en aval qui consommaient la sortie Avro
• A coûté à l'équipe l'observabilité du pipeline ; l'interface utilisateur de Spark et les métriques des travaux ne voyaient que le monde commençant à la zone de dépôt Avro, de sorte que les ingénieurs déboguant une ingestion lente ou échouée devaient aller chercher les journaux Lambda dans une partie complètement différente de la plateforme
• A impliqué de payer pour stocker à la fois les fichiers binaires d'origine et leurs copies Avro

Pourquoi Databricks et les sources de données personnalisées PySpark sont importants

Databricks était déjà la plateforme d'Eclipse Aerospace pour stocker et traiter les données de vol. L'avantage décisif a été la combinaison de l'exécution native Spark, de la source de données personnalisée PySpark et de Databricks Serverless. Databricks Serverless alloue et gère automatiquement le calcul à la demande, de sorte qu'Eclipse Aerospace n'a pas eu besoin de provisionner ou de maintenir une infrastructure dédiée en marche. La source de données personnalisée PySpark a permis d'encapsuler le décodeur Rust existant sans avoir à construire de connecteur JVM. Le décodage s'exécutant désormais dans Spark, la même plateforme fournit l'interface utilisateur de Spark, les métriques des travaux et les flux de travail Delta en aval, au lieu de confier l'étape d'ingestion critique à un pipeline Lambda séparé.

La majeure partie de l'ingénierie des données repose sur des formats de données courants, mais certaines des données les plus précieuses restent cachées dans des formats auxquels PySpark ne peut pas accéder directement, tels que HDF5 pour les données scientifiques et les normes industrielles telles que ARINC 429 pour les données aérospatiales. Databricks Serverless offre aux équipes un moyen flexible de libérer ces données à grande échelle, tandis que les implémentations basées sur Rust peuvent offrir des performances supérieures à une approche purement Python.

Les sources de données personnalisées PySpark vous permettent de définir un format de fichier lisible/inscriptible en pur Python, utilisable comme n'importe quelle autre source de données. Son principal avantage est la simplicité, permettant aux utilisateurs de définir des fonctions de lecture et d'écriture en pur Python, ainsi que de définir des caractéristiques importantes telles que la parallélisation. L'équipe d'Eclipse Aerospace est allé plus loin : en créant un paquet Python basé sur Rust, elle a obtenu deux fois les performances de référence d'une source de données pure Python similaire basée sur PyArrow. Par rapport à une implémentation Python native sans Arrow, elle a offert près de 20 fois les performances.

Le répertoire de code d'exemple suivant présente une source de données personnalisée pour ARINC 429. Le code d'accompagnement comprend des décisions de conception qui vont au-delà de la portée de cet article, ainsi que des tests de référence que vous pouvez exécuter vous-même.

Comment cela est devenu résoluble avec les sources de données personnalisées PySpark

Apache Spark 4.0 a intégré une API de source de données Python (mentionnée dans la documentation Databricks sous le nom de sources de données personnalisées PySpark) qui vous permet d'enregistrer un format personnalisé pour les données par lots ou en streaming comme suit :

Introduite dans Apache Spark 4.0 et disponible sur Databricks, l'API est généralement disponible à partir de Databricks Runtime 15.4 LTS et est également prise en charge sur l'environnement Serverless 2 et versions ultérieures. L'API étant implémentée en Python, elle ne nécessite pas de connecteur JVM ou de JAR séparé.

L'API est une interface simplifiée pour la création de sources de données natives, l'utilisateur devant implémenter quelques classes et les enregistrer avec :

Ces décodeurs se déclinent en deux variantes : Python natif et format Apache Arrow, ce dernier étant le choix le plus performant. Apache Arrow est un "format de mémoire en colonnes indépendant du langage pour les données plates et imbriquées", ce qui en fait une interface idéale entre différents langages partageant des objets de données. L'API en tire parti en permettant aux utilisateurs de renvoyer directement des objets pyarrow.RecordBatch, et Spark les consumera par lots plutôt que de matérialiser un InternalRow par ligne.

Cela a soulevé la question suivante : Python pouvait-il servir de couche de contrôle, tandis que Rust effectuait le travail lourd du décodage tout en maintenant le format Apache Arrow de bout en bout ?

Aperçu de l'architecture de la solution

L'architecture sous-jacente est un connecteur logique unique avec un chemin de décodage presque entièrement composé de code compilé, et une couche Python suffisamment légère pour n'apporter que des modifications minimes au SDK Rust existant d'Eclipse Aerospace. Cette couche Python légère est construite avec PyO3/Maturin en tant que paquet autonome, importé dans les pipelines et utilisé comme n'importe quelle autre source. Pour garantir les meilleures performances et la séparation des préoccupations, l'équipe a suivi quelques principes architecturaux.

Le décodeur ignore l'existence de Spark. Le crate Rust principal n'a aucune dépendance envers PyO3, Arrow ou Spark et accepte uniquement des octets pour renvoyer des enregistrements structurés. Cela a permis à Eclipse Aerospace de porter sa logique de décodage existante presque sans modification, le SDK n'ayant pas eu besoin d'être réécrit pour cela, mais simplement encapsulé.

Arrow est le seul élément qui traverse la frontière de l'interface de fonction étrangère (FFI), et il la traverse sans copie. La couche de liaison Rust construit un RecordBatch Arrow à l'aide de arrow-rs, puis le transmet à Python sous forme de pyarrow.RecordBatch via l'interface de données C d'Arrow. Ainsi, aucune étape de sérialisation n'a lieu, car aucun octet n'est ré-encodé et aucun objet Python par ligne n'est créé. Les tampons alloués par Rust sont les tampons lus par Python.

La couche PySpark est conçue pour être légère. Elle analyse les options, détermine le mode de partitionnement (un InputPartition par entrée dans l'exemple de référence) et définit le schéma. Dans les sources de données personnalisées PySpark, DataSource, DataSourceReader et leurs équivalents en streaming doivent tous être sérialisables afin que Spark puisse sérialiser (pickle) l'instance du lecteur et l'envoyer à un worker. En rendant la couche PySpark légère, l'instance DataSourceReader que Spark sérialise et envoie à un worker ne conserve que l'état Python de base (chemins de fichiers, un entier pour la taille du lot). L'objet Rust basé sur PyO3 est construit à l'intérieur de read(), une fois que le lecteur est déjà arrivé sur le worker. Il ne fait jamais lui-même partie des éléments sérialisés.

Le code natif est distribué sous forme de wheel Python, compilée avec Maturin. Sur Databricks, cette wheel constitue un artéfact de build, rattaché à un environnement de travail ou de notebook en tant que dépendance Python standard. L'implémentation étant compilée, la compatibilité des versions et des architectures nécessite une attention particulière, bien qu'un processus CI/CD adapté permette de gérer cet aspect, comme le montre le dépôt d'exemple.

Comment le connecteur s'exécute dans Databricks

Sur Databricks Serverless, le chemin d'exécution est direct :

1. Un fichier de télémétrie propriétaire est stocké dans le cloud.
2. Une lecture Spark sur Databricks appelle la source de données personnalisée PySpark enregistrée pour le format personnalisé.
3. Spark exécuté sur Databricks crée un lecteur sérialisable contenant le chemin d'entrée et les options de lecture, puis planifie les partitions d'entrée sur les workers.
4. Chaque worker construit le décodeur Rust dans read() et transmet les octets du fichier au décodeur compilé.
5. Rust renvoie des Arrow RecordBatches via l'interface C Data d'Arrow, évitant ainsi la sérialisation et la création d'objets Python par ligne.
6. Spark exécuté sur Databricks consomme les lots et écrit les données décodées dans des tables aval telles que Delta.
7. Les erreurs de décodage, les enregistrements erronés, les étapes de l'interface utilisateur Spark et les métriques de tâches restent visibles au sein du workflow Databricks.

C'est la différence clé par rapport à la conception précédente : le fichier d'origine est décodé lors de la lecture Spark au lieu d'être préalablement copié vers une zone de staging Avro intermédiaire.

Quel a été l'impact pour Eclipse Aerospace

En remplaçant le workflow Lambda et Avro en plusieurs étapes par une seule lecture native Spark, Eclipse Aerospace a rendu les données de vol disponibles plus rapidement tout en réduisant l'infrastructure, le stockage et les coûts opérationnels. Le nouveau connecteur de sources de données personnalisées PySpark natif pour Arrow intègre également le décodage dans Spark, améliorant la visibilité sur les performances du pipeline et permettant de gérer la logique de schéma et de qualité des données en un seul endroit.

La mise en œuvre de la source de données personnalisée a considérablement simplifié l'infrastructure d'Eclipse Aerospace, et les effets se font sentir dans quatre domaines distincts : la latence, les coûts, l'observabilité et le temps d'ingénierie.

Latence. La fusion d'un pipeline à plusieurs sauts (téléchargement, Lambda, décodage, écriture Avro, lecture Avro, chargement vers Delta) en une seule lecture native Spark supprime plusieurs sauts réseau et plusieurs points où un fichier patiente dans une file d'attente en attendant que l'étape suivante le récupère. Le nouveau connecteur peut traiter des enregistrements en moins d'une minute après le téléchargement du fichier, temps de démarrage compris, et parvient généralement à traiter en parallèle des lots volumineux de plusieurs gigaoctets en moins de deux minutes grâce au parallélisme.

Coût. La suppression du saut Lambda et de l'intermédiaire Avro a éliminé toute une couche d'infrastructure qu'Eclipse Aerospace devait exécuter, surveiller et financer. L'entreprise ne paie plus pour stocker deux copies de chaque vol (le fichier binaire d'origine et sa conversion Avro), ce qui représente un volume de stockage dupliqué important sur près de deux décennies d'historique de vol conservé. Par conséquent, la majorité des fichiers restent dans le stockage Glacier sur AWS, avec des décodages facilement reproductibles si les données d'origine s'avèrent nécessaires par la suite.

Observabilité. L'étape de décodage s'effectuant désormais directement au sein de Spark, elle apparaît dans l'interface utilisateur de Spark et dans les métriques de tâches standard, au même titre que n'importe quelle autre étape de travail. Eclipse Aerospace utilise la propriété "badRecordsPath", qui imite le chargeur CSV/JSON, pour capturer les données mal formatées, et maintient des pipelines de qualité des données exhaustifs qui capturent des statistiques directement depuis le décodeur.

Temps d'ingénierie. La logique de schéma était auparavant définie deux fois : une première fois dans le décodeur Lambda qui produisait du format Avro, et une seconde fois dans les tâches Spark aval qui le consommaient. Toute modification du format de message impliquait une mise à jour conjointe dans deux bases de code distinctes, avec le risque d'un décalage entre les deux définitions. Avec un connecteur unique et versionné, le schéma est défini une seule fois, aux côtés de la logique de décodage elle-même. Cela signifie également que le même modèle architectural, à savoir un connecteur Python léger encapsulant un décodeur Rust, est désormais réutilisable pour les autres formats propriétaires pris en charge par Eclipse Aerospace, évitant ainsi de devoir tout réinventer pour chaque format.

L'expérience d'Eclipse Aerospace démontre comment des formats de données spécialisés peuvent être intégrés dans Databricks sans nécessiter la création d'une plateforme d'ingestion dédiée. En maintenant le décodeur en Rust, en utilisant Python comme couche d'intégration et en faisant transiter les enregistrements par Apache Arrow, la source de données personnalisée lit les fichiers d'origine directement dans Spark tout en préservant la compatibilité avec la souplesse de Databricks Serverless.

Pour les équipes travaillant avec des formats propriétaires ou spécifiques à leur secteur, le modèle réutilisable est simple. Les ingénieurs peuvent conserver la logique de décodage propre à leur domaine près du format source, l'exposer via une source de données personnalisée PySpark et laisser Spark gérer le partitionnement, l'exécution, l'observabilité et le traitement aval.

Vous souhaitez commencer ? Jetez un œil aux ressources suivantes :

• PySpark sur Databricks
• Sources de données personnalisées PySpark
• Se connecter au compute sans serveur

Foire aux questions

Qu'est-ce qu'une source de données personnalisée PySpark ?

Une source de données personnalisée PySpark est une source de données Spark définie par l'utilisateur et implémentée en Python. Elle permet aux équipes de lire des formats de fichiers spécialisés dans le cadre d'un workflow Spark standard, sans avoir à développer de connecteur JVM distinct.

Quel problème d'ingestion de données un connecteur de source de données personnalisée PySpark natif pour Arrow résout-il ?

Le connecteur de source de données personnalisée PySpark natif pour Arrow intègre la télémétrie de vol propriétaire dans Databricks sans s'appuyer sur un pipeline à plusieurs étapes, un compute dédié fonctionnant en continu ou des copies Avro en double des fichiers d'origine.

Qu'est-ce que Databricks Serverless ?

Databricks Serverless est un service de calcul entièrement géré qui vous permet d'exécuter des charges de travail sans avoir à provisionner ni gérer d'infrastructure au sein de votre compte cloud. Databricks alloue, met à l'échelle et optimise automatiquement les ressources de calcul pour vous.

Comment Databricks Serverless se connecte-t-il au connecteur de source de données personnalisée PySpark natif pour Arrow ?

Le connecteur utilise les sources de données personnalisées PySpark pour exécuter la logique de lecture de fichiers au sein de Spark sur un environnement de calcul sans serveur. Python gère le flux de contrôle, tandis que Rust et Apache Arrow prennent en charge le décodage haute performance et le transfert de données. Cela permet de lire directement les fichiers propriétaires sans nécessiter de compute dédié fonctionnant en continu ni de pipeline Avro intermédiaire. Les sources de données personnalisées PySpark sont disponibles dans Databricks Runtime 15.4 LTS et versions ultérieures, ou dans l'environnement sans serveur version 2.

(Cet article de blog a été traduit à l'aide d'outils basés sur l'intelligence artificielle) Article original