Eclipse Aerospace が Databricks を使用してフライトデータの取り込みを簡素化

航空宇宙、製造、ライフサイエンス、その他のデータ集約型業界の組織が業務を近代化するにつれて、より多くのセンサー、テレメトリー、科学データをデータプラットフォームに取り込んでいます。しかし、このデータの多くは、主要な分析エンジンが直接読み取ることができない特殊または独自のフォーマットで存在するため、大規模での利用が依然として困難なままです。これにより、実践的な業界の課題が生じています。チームは、遅くメモリを大量に消費するカスタム変換の構築、専用コネクタの維持、またはデータの分析前に追加のデコードステージの追加のいずれかを選択する必要があります。

PySparkカスタムデータソースは、よりシンプルな解決策を提供します。これにより、チームはゼロから別のコネクタを構築することなく、Pythonを使用して特殊なフォーマットをSparkに取り込むことができます。非常に軽いジェット機のメーカー兼運航会社であるEclipse Aerospaceにとって、Databricksとのパートナーシップは、コストのかかる常時稼働のインフラストラクチャを追加することなく、約20年分の独自のフライトデータをサービスセンターやエンジニアにとってよりアクセスしやすいものにする方法を提供しました。DatabricksはサーバーレスデータプラットフォームとSpark統合を提供し、Eclipse Aerospaceは既存のフライトデータデコーダーと航空宇宙の専門知識を提供しました。

Eclipse AerospaceとDatabricksは協力して、フライトデータからアナリティクスまでのより直接的なパスを開発し、Databricks Serverlessの柔軟性を維持しながら、以前のマルチステージプロセスのレイテンシ、重複、および運用上のオーバーヘッドを削減しました。

Eclipse Aerospaceのデータ取り込みの課題

Eclipse Aerospaceは、合計3TBを超える約20年分のフライトデータを保存するためにDatabricksに依存しています。次のような複数のソースから取得および検証された100万時間以上のフライト時間のデータで、サービスセンターとエンジニアをサポートしています：ほぼリアルタイムの衛星テレメトリー、完全なフライト履歴の大規模なバルクアップロード、および外部データソース。すべてのテレメトリーは、RustやScalaなど複数の言語ですでに開発されているカスタムデコーダーを必要とする独自のファイル形式で保存されます。

PySparkカスタムデータソースがDatabricks Runtime 15.4 LTSで利用可能になる前、Eclipse Aerospaceチームには、独自のファイルタイプをデコードしてDatabricksにロードするための3つの実行可能な選択肢がありました：

1. ファイルをバイナリ型としてロードし、Pythonユーザー定義関数（UDF）を使用して分割する
2. 専用のコンピュート上でScalaのカスタムデコーダーを使用する
3. Databricksにロードする前にファイルを中間ファイルタイプにデコードする

単純なカスタムファイルタイプの場合、UDFは自然な選択肢です。パフォーマンスと引き換えに可読性を高めます。ただし、メッセージフォーマットの長さにバリエーションがある場合、UDFは長いメッセージでメモリ不足になる現実的なリスクがあり、使用できませんでした。既存のScalaデコーダーを使用してネイティブなData Source v1/v2を実装することはできましたが、Databricksで使用するには専用のコンピュートが必要でした。ユーザーはデータを素早くアップロードして表示できることを期待しており、これにはこのコンピュートを常時稼働させておく必要があるため、他のサーバーレスオプションよりも大きなコストが発生することになります。これにより、チームには、AWS Lambda上のRustベースのデコーダーを介して、ファイルを中間ファイル形式（この場合はAvro）にデコードするという最後のオプションが残されました。

このソリューションは低コストでしたが、次のような問題がありました：

• マルチホップの取り込みパターンによるレイテンシが発生した
• Lambdaデコーダーと、Avro出力を消費するダウンストリームのSparkジョブの間でスキーマロジックが重複した
• チームのパイプラインの観測性が損なわれました。Spark UIとジョブメトリックはAvroランディングゾーンから始まる世界しか認識できなかったため、低速または失敗した取り込みをデバッグするエンジニアは、プラットフォームのまったく異なる部分にあるLambdaログを探しに行く必要がありました。
• 元のバイナリファイルとそのAvroコピーの両方を保存するための費用を支払う必要があった

DatabricksとPySparkカスタムデータソースが重要な理由

DatabricksはすでにEclipse Aerospaceのフライトデータを保存および処理するためのプラットフォームでした。決定的な利点は、Sparkネイティブの実行、PySparkカスタムデータソース、およびDatabricks Serverlessの組み合わせでした。Databricks Serverlessはオンデマンドのコンピュートを自動的にプロビジョニングおよび管理するため、Eclipse Aerospaceは専用のインフラストラクチャをプロビジョニングしたり稼働させ続けたりする必要がありませんでした。PySparkカスタムデータソースにより、JVMコネクタを構築することなく、既存のRustデコーダーをラップすることが可能になりました。デコードがSpark内で実行されるようになったため、重要な取り込みステップを別のLambdaパイプラインに引き渡すのではなく、同じプラットフォームがSpark UI、ジョブメトリック、およびダウンストリームのDeltaワークフローを提供します。

ほとんどのデータエンジニアリングは一般的なデータフォーマットに依存していますが、最も価値のあるデータの一部は、科学データのためのHDF5や、航空宇宙データのためのARINC 429などの業界標準など、PySparkが直接アクセスできないフォーマットの中に隠されたままになっています。Databricks Serverlessは、チームが大規模でそのデータのロックを解除する柔軟な方法を提供し、Rustベースの実装は純粋なPythonアプローチを超えるパフォーマンスを提供できます。

PySparkカスタムデータソースを使用すると、純粋なPythonで読み取り/書き込み可能なファイル形式を定義でき、他のデータソースと同様に使用できます。その主な利点はシンプルさであり、ユーザーは純粋なPythonでの読み取りと書き込みの関数を定義できるだけでなく、並列化などの重要な特性も定義できます。Eclipse Aerospaceチームはこれをさらに一歩進めました。RustベースのPythonパッケージを構築することにより、PyArrowベースの同様の純粋なPythonデータソースの2倍のベンチマークパフォーマンスを実現しました。Arrowを使用しないネイティブなPython実装と比較して、約20倍のパフォーマンスを実現しました。

次のコード例のリポジトリでは、ARINC 429のカスタムデータソースについて説明しています。付属のコードには、この記事の範囲外の設計上の決定事項や、自分で実行できるベンチマークテストが含まれています。

PySparkカスタムデータソースでこれがどのように解決可能になったか

Apache Spark 4.0には、バッチデータやストリーミングデータのカスタムフォーマットを登録できるPythonデータソースAPI（DatabricksドキュメントではPySparkカスタムデータソースと称されています）が搭載されています：

Apache Spark 4.0で導入され、Databricksで利用可能なこのAPIは、Databricks Runtime 15.4 LTS以降で一般提供（GA）されており、Serverレス環境2以降でもサポートされています。APIはPythonで実装されているため、個別のJVMコネクタやJARは必要ありません。

APIはネイティブデータソースを作成するための簡素化されたインターフェイスであり、ユーザーはいくつかのクラスを実装して次のように登録します：

これらのデコーダーには、ネイティブPythonとApache Arrowフォーマットの2種類があり、後者はパフォーマンスの面でより優れた選択肢です。Apache Arrowは、「フラットデータおよびネストされたデータのための言語非依存のカラム型メモリフォーマット」であり、データオブジェクトを共有する異なる言語間の理想的なインターフェイスとなっています。APIはこの利点を活かし、ユーザーがpyarrow.RecordBatchオブジェクトを直接返すことを可能にし、Sparkは1行ごとに1つのInternalRowを具体化するのではなく、バッチとしてそれらを消費します。

これにより、次のような疑問が生じました：Pythonを制御レイヤーとして機能させ、Rustにデコードの重い処理を行わせながら、Apache Arrowフォーマットを全体を通して維持することはできないでしょうか？

ソリューションアーキテクチャの概要

これの背後にあるアーキテクチャは、デコードパスがほぼ完全にコンパイルされたコードである単一の論理コネクタであり、Eclipse Aerospaceの既存のRust SDKに変更をほとんど加えないほどスリムなPythonレイヤーです。このスリムなPythonレイヤーは、スタンドアロンパッケージとしてPyO3/Maturinで構築され、パイプラインにインポートされ、他のソースと同じように使用されます。最高のパフォーマンスと関心の分離を確保するために、チームはいくつかのアーキテクチャ原則に従いました。

デコーダーはSparkの存在を知りません。コアのRustクレートには、PyO3、Arrow、またはSparkへの依存関係がなく、バイトを受け取るだけで、構造化されたレコードを返します。これにより、SDKをこれ用に書き直す必要がなく、ラップするだけで済んだため、Eclipse Aerospaceは既存のデコードロジックをほとんど変更せずに移植することができました。

Arrowは外部関数インターフェイス（FFI）境界を越える唯一のものであり、コピーなしで越えます。Rustのグルーレイヤーは、arrow-rsを使用してArrowのRecordBatchを構築し、Arrow C Data Interfaceを介してpyarrow.RecordBatchとしてPythonに渡します。これを行う際、バイトが再エンコードされたり、行ごとのPythonオブジェクトが構築されたりしないため、シリアル化のステップはありません。Rustが割り当てたバッファは、Pythonが読み取るバッファです。

PySparkレイヤーは意図的に軽量に設計されています。オプションを解析し、パーティショニングの方法を決定し（参考例では入力ごとに1つのInputPartition）、スキーマを定義します。PySparkのカスタムデータソースでは、DataSource、DataSourceReader、およびそれらのストリーミング版はすべてシリアライズ可能である必要があり、そのためSparkはリーダーインスタンスをピックルしてワーカーに送信できます。PySparkレイヤーを軽量に保つことで、Sparkがピックルしてワーカーに送信するDataSourceReaderインスタンスは、プレーンなPythonの状態（ファイルパス、バッチサイズを表すint）のみを保持します。PyO3ベースのRustオブジェクトは、リーダーがすでにワーカーに到着した後のread()内で構築されます。ピックルされるデータの一部になることは決してありません。

ネイティブコードは、MaturinでビルドされたPythonホイールとして提供されます。Databricksでは、そのホイールはビルド成果物であり、通常のPython依存関係としてジョブまたはノートブック環境にアタッチされます。実装がコンパイルされているため、バージョンとアーキテクチャの互換性には注意が必要ですが、適切なCI/CDプロセスがあれば、サンプルリポジトリで示されているようにこれを管理できます。

Databricksでのコネクタの実行方法

Databricks Serverlessでは、ランタイムパスはシンプルです。

1. 独自のテレメトリファイルがクラウドストレージに保存されます。
2. Databricks Sparkの読み取りにより、カスタムフォーマット用に登録されたPySparkカスタムデータソースが呼び出されます。
3. Databricks上で実行されるSparkは、入力パスと読み取りオプションを含むシリアライズ可能なリーダーを作成し、ワーカーに入力パーティションをスケジュールします。
4. 各ワーカーは、read()内でRustデコーダーを構築し、ファイルバイトをコンパイル済みデコーダーに渡します。
5. Rustは、Arrow C Data Interfaceを介してArrow RecordBatchesを返し、シリアライズや行ごとのPythonオブジェクトを回避します。
6. Databricks上で実行されるSparkはバッチを処理し、デコードされたデータをDeltaなどの下流テーブルに書き込みます。
7. デコードエラー、不正なレコード、Spark UIのステージ、およびジョブのメトリクスは、Databricksワークフロー内で確認できます。

これが従来のデザインとの大きな違いです。元のファイルは、最初に中間層のAvroランディングゾーンにコピーされるのではなく、Sparkの読み取りの一部としてデコードされます。

Eclipse Aerospaceにとってどのような影響があったか

マルチステージのLambdaとAvroのワークフローを単一のSparkネイティブの読み取りに置き換えることで、Eclipse Aerospaceはフライトデータをより迅速に利用可能にし、インフラストラクチャ、ストレージ、および運用のオーバーヘッドを削減しました。また、新しいArrowネイティブのPySparkカスタムデータソースコネクタにより、デコードがSparkに組み込まれ、パイプラインのパフォーマンスの可視性が向上し、スキーマとデータ品質のロジックを1か所で管理できるようになりました。

カスタムデータソースの実装により、Eclipse Aerospaceのインフラストラクチャは大幅にシンプルになり、その効果はレイテンシ、コスト、可観測性（オブザーバビリティ）、エンジニアリング時間の4つの明確な領域に表れています。

レイテンシ。マルチホップのパイプライン（アップロード、Lambda、デコード、Avroの書き込み、Avroの読み取り、Deltaへのロード）を単一のSparkネイティブの読み取りに統合することで、複数のネットワークホップや、次のステージが処理するのをファイルがキューで待機するポイントがいくつか削除されます。新しいコネクタは、起動時間を含めてファイルのアップロードから1分以内にレコードを処理でき、並列処理のおかげで、一般的に数ギガバイトに及ぶ大規模なバッチを2分未満で並列処理できます。

コスト。LambdaホップとAvroの中間層を削除したことで、Eclipse Aerospaceが実行、監視、維持のために費用を支払う必要があったインフラストラクチャの階層全体が削減されました。すべてのフライトの2つのコピー（元のバイナリとそのAvro変換）を保存するための費用を支払う必要がなくなり、これは約20年間にわたる蓄積されたフライト履歴において、重複するストレージの大幅な削減を意味します。これにより、ファイルの大部分はAWSのGlacierストレージに保持され、元のデータが再び必要な場合には簡単に再現可能なデコードを行うことができます。

可観測性（オブザーバビリティ）。デコードステップがSpark内で直接行われるようになったため、他のジョブステージと同様に、Spark UIや標準のジョブメトリクスに表示されます。Eclipse Aerospaceでは、CSV/JSONローダーを模したカスタムの「badRecordsPath」を使用して不正な形式のデータをキャプチャし、デコーダーから直接統計情報を取得する拡張性の高いデータ品質パイプラインを維持しています。

エンジニアリング時間。スキーマロジックは以前は2回定義されていました。1回目はAvroを生成するLambdaデコーダーで、2回目はそれを消費する下流のSparkジョブです。メッセージ形式を変更するということは、2つの異なるコードベースで両方を更新することを意味し、2つの定義が乖離するリスクがありました。単一のバージョン管理されたコネクタを使用することで、スキーマはデコードロジックのすぐ隣に1回だけ定義されます。これは、Eclipse Aerospaceがサポートする他の独自形式に対しても、形式ごとに再発明するのではなく、薄いPython DataSourceがRustデコーダーをラップする同じアーキテクチャパターンを再利用できることを意味します。

Eclipse Aerospaceの経験は、独自のデータ形式を別途のインングestionプラットフォームを構築することなくDatabricksに統合できることを示しています。デコーダーをRustで維持し、統合レイヤーとしてPythonを使用し、Apache Arrowを通じてレコードを渡すことで、カスタムデータソースは元のファイルをSparkで直接読み取り、Databricks Serverlessの柔軟性と互換性を維持します。

独自形式や業界固有の形式を扱うチームにとって、再利用可能なパターンはシンプルです。エンジニアはドメイン固有のデコードロジックをソース形式の近くに配置し、PySparkカスタムデータソースを介して公開し、パーティショニング、実行、可観測性、および下流の処理をSparkに処理させることができます。

開始方法については、以下をご覧ください。

• DatabricksでのPySpark
• PySparkカスタムデータソース
• サーバーレスコンピュートへの接続

よくある質問

PySparkカスタムデータソースとは何ですか？

PySparkカスタムデータソースは、Pythonで実装されたユーザー定義のSparkデータソースです。チームは、別のJVMコネクタを構築することなく、標準的なSparkワークフローの一部として特殊なファイル形式を読み取ることができます。

ArrowネイティブのPySparkカスタムデータソースコネクタはどのようなデータ取り込みの問題を解決しますか？

ArrowネイティブのPySparkカスタムデータソースコネクタは、マルチステージパイプライン、専用の常時稼働コンピュート、または元のファイルの重複するAvroコピーに依存することなく、独自のフライトテレメトリをDatabricksに取り込みます。

Databricks Serverlessとは何ですか？

Databricks Serverlessは、クラウドアカウント内のインフラストラクチャをプロビジョニングまたは管理することなくワークフローを実行できる、完全管理型のコンピュートサービスです。Databricksが自動的にコンピュートリソースを割り当て、スケーリングし、最適化します。

Databricks Serverlessは、ArrowネイティブのPySparkカスタムデータソースコネクタにどのように接続しますか？

コネクタは、Pysparkカスタムデータソースを使用して、サーバーレスコンピュート上のSpark内でファイル読み取りロジックを実行します。Pythonが制御フローを処理し、RustとApache Arrowが高性能なデコードとデータ転送を処理します。これにより、専用の常時稼働コンピュートや中間層のAvroパイプラインを使用せずに、独自のファイルを直接読み取ることができます。PySparkカスタムデータソースは、Databricks Runtime 15.4 LTS以降、またはサーバーレス環境バージョン2で利用できます。

(このブログ記事はAI翻訳ツールを使用して翻訳されています) 原文記事