PythonデータソースAPIの一般提供を開始

Databricks Runtime (DBR) 15.4 LTS 以降で利用可能な、Apache Spark™ 4.0 向け PySpark データソース API の一般提供 (GA) を開始したことをお知らせします。この強力な機能を使うと、開発者は純粋なPythonだけでSparkのカスタムデータコネクタを構築できます。これにより、外部データソースや Spark ネイティブではないデータソースとの統合が簡素化され、データ パイプラインと機械学習ワークフローの新たな可能性が広がります。

これが重要な理由

今日では、データはさまざまなソースから取り込まれています。構造化されたデータもあれば、非構造化データもあり、画像や動画のようなマルチモーダルデータも存在します。Sparkは、Delta、Iceberg、Parquet、JSON、CSV、JDBCなどのデータソースv1 (DSv1) およびデータソースv2 (DSv2) の標準フォーマットをネイティブでサポートしています。しかし、Googleスプレッドシート、REST API、HuggingFaceデータセット、Xのツイート、独自の社内システムなど、他の多くのソースに対しては組み込みのサポートを提供していません。DSv1/DSv2は、技術的にはこれらのソースを実装するために拡張できますが、そのプロセスは過度に複雑であり、軽量なユースケースでは多くの場合不要です。

ご自身のユースケースでこれらの他のカスタムデータソースからの読み取りや書き込みが必要な場合や、machine learningのユースケース向けのETLパイプラインでモデルをトレーニングするためにこのデータを使用する必要がある場合はどうでしょうか？Python データソース API が埋めるのは、まさにそのギャップです。

このブログでは、PySparkでカスタムデータソースを作成する方法について解説します。このAPIを使用すると、Sparkに組み込まれていない多様なデータセットを、特定のユースケースに合わせてデータ処理パイプラインに簡単に取り込むことができます。また、カスタムデータソースのいくつかの例についても見ていきましょう。しかし、まずはその理由と内容について理解しましょう。

Python データソース API とは何ですか?

Python での記述が好まれ、pip を使ってパッケージを簡単にインストールできることから、Python データソース API は、Python を使って Spark 用のカスタムリーダーとライターを簡単に構築できるようにします。この API は、あらゆるデータソースへのアクセスを可能にします。これにより、これまでカスタムコネクタに必要だった複雑な DSv1 や DSv2 の開発、あるいは Spark の内部構造に関する知識が不要になります。

主な機能とメリット

Pythonデータソースを使用することには、いくつかの主要な利点があります。

1. 純粋な Python による実装

まず、Python らしい使い心地について考えてみましょう。Python データソース API は、より複雑な JVM ベースのコネクタ開発の障壁を取り除きます。Pythonで多数のデータソースを使って複雑なETLパイプラインを構築しているデータエンジニアは、Sparkの内部の複雑さに踏み込むことなく、既存のPythonスキルを活用してカスタムコネクタを作成できるようになりました。

2. バッチ処理とストリーミング処理の両方をサポート

この API はバッチ読み取りとストリーミング読み取りの両方をサポートしており、複数のデータアクセス パターンを処理するコネクタを構築できます。

• バッチ読み取り: API、データベース、またはその他のソースから 1 回の操作でデータを取得します
• ストリーム読み取り: イベントベースまたはリアルタイムのソースからデータを継続的に取り込みます
• バッチ書き込みとストリーム書き込み: Declarative Pipeline シンクを含め、データシンクに書き込みます。

3. SQLからのアクセシビリティ

Spark SQLでサポートされているデータソースと同様に、Spark SQLからカスタムのPythonデータソースにも簡単にアクセスできます。データソースがDataFrameとして読み込まれると、一時ビューまたは永続的なUnity Catalogのマネージドテーブルとして保存できます。これにより、ダウンストリームのSQLアナリティクスにカスタムデータソースを組み込むことができます。

4. 外部サービスとの統合の簡素化

DataFrame APIを使い、APIキー、エンドポイント、その他の設定といったカスタムオプションを渡すことで、外部システムに簡単に接続できます。これにより、コネクタの動作を完全に制御できます。詳細については、以下の実例をご覧ください。

5. コミュニティ主導のコネクタ エコシステム

プレビューリリース以来、コミュニティはすでにPythonデータソースAPIを使用して、価値あるコネクタの構築を開始しています。

• コネクタの例: REST API、さまざまなCSV、その他のためのリファレンス実装（GitHubリポジトリ）
• HuggingFaceコネクタ: HuggingFaceのデータセットに直接アクセス（GitHubリポジトリ）

6. 複雑さのないスピード

Python データソース API は、使いやすさだけでなく、速度も考慮して設計されています。高速データ処理用に最適化されたインメモリデータ形式である Apache Arrow を基盤としています。これにより、カスタムコネクタと Spark 間でデータを最小限のオーバーヘッドで移動でき、取り込みと書き込みがはるかに高速になります。

Python データソースの実際のユースケース

Python データソースを使用してパイプラインにデータを取り込むための、実際のユースケースを見ていきましょう。

カスタム API 統合

まず、REST APIを統合することでどのようなメリットがあるかを見ていきましょう。

多くのデータエンジニアリングチームは、REST API からデータを取得し、それを下流の変換で使用するためのカスタムコネクタを構築しています。データを取得してディスクやメモリに保存し、それをSparkにロードするためのカスタムコードを作成する代わりに、PythonデータソースAPIを使用することでこれらのステップを省略できます。

カスタムデータソースを使用すると、API からデータを直接 Spark DataFrame に読み込むことができ、中間ストレージは必要ありません。たとえば、REST API 呼び出しから完全な出力を取得し、Spark に直接読み込む方法は次のとおりです。

REST API データソースとその完全な実装は、参照例としてこちらにあります。

Unity Catalog 統合

2つ目の統合はデータカタログです。企業では、データの一元的なガバナンスとセキュリティを実現するために、AI とデータ資産を中央リポジトリに格納する、Unity Catalog のようなデータカタログに注目する開発者が増えています。この2つ目のトレンドが続くにつれて、データパイプラインのジョブは、これらのデータ資産に対して、安全かつ管理された方法で読み書きができる必要があります。

これらのカスタムデータソースから直接データを読み込んで Unity Catalog テーブルに書き込むことで、あらゆるソースのデータにガバナンス、セキュリティ、検出可能性をもたらすことができます:

この統合により、特殊なソースからのデータを Unity Catalog を通じて適切に管理および保護できるようになります。

機械学習パイプラインの統合

3つ目の統合は、機械学習の外部データセットとの連携です。データ サイエンティストは Python データソース API を使用して、特殊な機械学習 (ML) データセットやモデル リポジトリに直接接続します。HuggingFaceには、古典的なMLモデルのトレーニングとテストのために厳選されたデータセットが膨大にあります。

このデータセットをSpark DataFrameとして取得するには、HuggingFaceコネクタを使用できます。このコネクタは、PythonデータソースAPIの能力を活用して、データパイプラインに統合できるMLアセットを簡単に取得します。

取得後、適切な Spark DataFrame を関連する machine learningアルゴリズムで使用して、モデルをトレーニングする、テスト、評価ができます。シンプル

その他の例については、HuggingFace DataSource Connector をご覧ください。

カスタムソースによるストリーム処理

さらに4つ目の統合ポイントとして、ストリーミングデータソースも、ストレージ上の静的ソースと同様に、日々のETLパイプラインの一部となっています。継続的にデータを取り込むカスタムソースを使用して、ストリーミング アプリケーションを構築することもできます。

これは、OpenSky Network APIから航空機のリアルタイム追跡データをストリーミングするためのカスタムSparkデータソースのコードスニペットです。コミュニティ主導の受信機ネットワークであるOpenSky Networkは、航空交通監視データを収集し、研究者や愛好家向けにオープンデータとして提供しています。このストリーミングカスタムデータソースの完全な実装を確認するには、こちらのGitHubソースをご覧ください。

宣言的パイプラインの統合

最後に、データエンジニアは Python データソースを 宣言的パイプライン統合と簡単に統合できます。

カスタムデータソースからの読み取り

宣言型パイプラインでカスタムデータソースからデータを取り込む方法は、通常の Databricks Jobs と同じです。

カスタムデータソース経由での外部サービスへの書き込み

このブログの例では、Alex Ott が、新しい Declarative Pipelines Sinks API を使用して、外部の Delta テーブルや Kafka などの他のストリーミング ターゲットを指すシンク オブジェクトを組み込みデータソースとして使用する方法を紹介しています。

ただし、Python データソースとして実装されたカスタムシンクに書き込むこともできます。以下のコードのカスタム Python データソースでは、sink API を使用してシンクを作成し、それを「sink」オブジェクトとして使用します。定義して開発すれば、フローを追加できます。このコードの完全な実装はこちらでご覧いただけます。

独自のカスタムコネクタの構築

PythonデータソースAPIを使用してカスタムコネクタを構築するには、次の4つのステップに従ってください。

1. Spark 4.0またはDatabricks Runtime 15.4 LTS以降があることを確認してください: PythonデータソースAPIは、DBR 15.4 LTS以降で利用可能であるか、Databricks Serverless Generic Computeを使用します。
2. 実装テンプレートを使用する: pyspark.sql.datasourceモジュールのベースクラスを参照してください。
3. コネクタの登録: Databricksワークスペースでコネクタを検出可能にします。

コネクタを使用する: コネクタは、組み込みのデータソースとまったく同じように使用できます。

PythonデータソースAPIを使用した顧客の成功事例

Shell社では、データエンジニアがApache KafkaのようなSpark組み込みソースのデータと、REST APIやSDKでアクセスする外部ソースのデータを組み合わせる必要が頻繁にありました。その結果、場当たり的でオーダーメイドのカスタムコードが生まれ、保守が難しく、チーム間での一貫性もありませんでした。シェルの最高デジタル技術責任者は、それが時間の無駄となり、複雑さを増したと指摘しました。

“私たちは、ストリーミングのユースケース向けのものも含め、多くの優れた REST API を作成しています。自分たちで面倒な定型コードをすべて書くのではなく、それらを Databricks でデータソースとしてそのまま使用したいのです。”
シェル デジタル技術部門チーフアドバイザー Bryce Bartmann 氏

その優れた REST API を活用するため、シェルのデータエンジニアが採用したソリューションは、新しい Python カスタムデータソース API を使用してREST API をデータソースとして実装することでした。これにより、開発者は APIs やその他の非標準ソースをファーストクラスの Spark データソースとして扱うことができるようになりました。オブジェクト指向の抽象化により、カスタムロジックをクリーンに組み込むことが容易になり、煩雑なグルーコードは不要になります。

まとめ

まとめると、PySpark の Python Data Source API を使用すると、Python 開発者は Spark 内部の深い知識がなくても、使い慣れた優れた Python を使用して、シンプルさとパフォーマンスを両立させながら、カスタムデータを Apache Spark™ に取り込むことができます。REST APIへの接続、HuggingFaceのような機械学習データセットへのアクセス、ソーシャルプラットフォームからのデータストリーミングのいずれの場合でも、このAPIはクリーンでPythonicなインターフェースによってカスタムコネクタの開発を簡素化します。これにより、JVM ベースのコネクタの複雑さを回避し、データチームが PySpark と SQL で直接ソースを構築、登録、使用できる柔軟性が得られます。

バッチおよびストリーミングのワークロードをサポートし、Unity Catalog と統合することで、この API は外部データのガバナンスとアクセシビリティを確保します。リアルタイムの ML パイプラインから分析やデータ インジェストまで、Python Data Source API は Spark を拡張可能なデータプラットフォームへと変えます。

今後のロードマップ

PythonデータソースAPIは進化を続けており、私たちは今後予定されているいくつかの機能強化に期待しています。

• カラムプルーニングとフィルタプッシュダウン: フィルタリングと列の選択をソースに近い場所で行うことでデータ転送を最適化する、より高度な機能。
• カスタム統計のサポート: クエリー プランニングと最適化を改善するために、コネクタがソース固有の統計情報を提供できるようにします。
• 優れた可観測性とデバッグの容易さ: コネクタ開発とトラブルシューティングを簡素化する、強化されたロギングツール
• 拡張されたサンプル ライブラリ: 一般的なデータソースと使用パターンのための、より多くのリファレンス実装。
• パフォーマンスの最適化: シリアル化のオーバーヘッドを削減し、スループットを向上させるための継続的な改善

今すぐお試しください

PythonデータソースAPIは、 Databricks Runtime 15.4 LTS 以降および サーバーレス環境 を含む Databricks Intelligence Platform で、本日より一般提供されます。

カスタムコネクタの構築を開始して、あらゆるデータソースをデータレイクハウスに統合しましょう！また、Data + AI Summit 2025で発表されたこれらの講演を聴いて、他の人がどのようにPython DataSourceを実装したかを学びましょう:

• 障壁を打ち破る: Python によるカスタム Apache Spark™ 4.0 データソースコネクタの構築
• PySpark 4.0 を使用したカスタム PySpark ストリーム リーダーの作成
• 新しいPythonソースAPIでデータインジェストとエグレスを簡素化

詳細は、PythonデータソースAPIの個別デモについてはお問い合わせいただくか、利用を開始するにはドキュメントをご覧ください。

ドキュメントとリソース

• Python データソース ドキュメント
• 宣言型パイプライン
• 宣言型パイプラインのシンク
• コネクタの例のリポジトリ
• HuggingFace Connector
• サイバーセキュリティのユースケースのための、最新の宣言型パイプライン機能の効率的な活用
• Delta Live Tables レシピ: Unity Catalog サービス資格情報を使用した Azure Event Hubs からのデータ消費
• PySpark 向けのニュースデータソース API
• HuggingFace AI データセット向けの Apache Spark™ データソース