Spark データ ソース

Spark データ ソース API を使用すると、Azure Databricksから直接外部データベースの読み取りと書き込みを行うことができます。 Spark エンジンの完全な柔軟性が必要な場合、ソースでネイティブ クエリを実行する場合、または外部システムへの書き込みアクセスが必要な場合にのみ使用します。 一般に、Azure Databricksでは、Spark または SQL クエリの自動プッシュダウンを使用した、管理された読み取り専用アクセスが推奨されます。 「クエリフェデレーションとは」を参照してください。

Spark データ ソース API には、接続、クエリの実行、スキーマ検出に固有の動作があります。

  • プライマリ ワークロードとその後の Spark 変換は、Azure Databricks Spark クラスターで実行されます。
  • query オプションを使用する場合、指定した SQL ステートメントは外部データ ソースで完全に実行されます。 Spark は、クエリ文字列に対して変換プッシュダウンを実行せずに結果をフェッチします。
  • 接続には、Azure Databricksバンドルされたコネクタ、ユーザー指定の JDBC ドライバー、または PySpark カスタム データ ソースのいずれかが必要です。
  • Spark は、外部データベース テーブルからスキーマを自動的に読み取り、その型を Spark SQL 型にマップします。

バンドルされたコネクタを使用する

Databricks Runtime には、一般的なデータ ソース用に最適化されたコネクタが含まれています。 完全な一覧については、 サポートされているバンドルされたコネクタ を参照してください。

バンドルされたコネクタでは、完全な JDBC URL 文字列ではなく、 host と port を個別のオプションとして使用します。

パススルー クエリを使用してデータを読み取る

query オプションを使用すると、データが Spark に到達する前に、ソース データベースでフィルターと結合ロジックが確実に実行されます。 ビューを介した自動クエリ プッシュダウンと Unity カタログのアクセス許可委任による管理された読み取りアクセスについては、代わりに リモート クエリ を検討してください。

df = (spark.read
  .format("sqlserver")
  .option("host", "<your-sql-server-instance>.database.windows.net")
  .option("user", dbutils.secrets.get(scope="<scope>", key="<user>"))
  .option("password", dbutils.secrets.get(scope="<scope>", key="<password>"))
  .option("database", "<database-name>")
  .option("query", "SELECT id, name FROM users WHERE active = 1")
  .load())

データを書き込む

データの書き込み方法を制御する .mode() を持つ書き込みモードを指定します。 appendを使用して既存のテーブルに行を追加するか、overwriteを使用してその内容を置き換えます。

(df.write
  .format("sqlserver")
  .mode("overwrite")
  .option("host", "<your-sql-server-instance>.database.windows.net")
  .option("user", dbutils.secrets.get(scope="<scope>", key="<user>"))
  .option("password", dbutils.secrets.get(scope="<scope>", key="<password>"))
  .option("database", "<database-name>")
  .option("dbtable", "<table-name>")
  .save())

JDBC UC 接続を使用する

ソース固有のコネクタがバンドルされていない場合、または特定の JDBC ドライバー バージョンを使用する場合は、JDBC Unity カタログ接続を使用します。 これにより、資格情報の管理を一元化し、独自の JDBC ドライバーを使用できます。

JDBC Unity カタログ接続には、バンドルされたコネクタまたは生の JDBC ドライバーを直接使用するよりも、いくつかの利点があります。 JDBC Unity カタログ接続を使用すると、次のことができます。

  • JDBC をサポートする任意のデータベースに対して、独自の JDBC ドライバー JAR を使用します。
  • 接続を 1 回作成し、サーバーレス、Standard、および専用クラスター間で再利用します。
  • Unity カタログ接続オブジェクトを使用して、データ ソースへの管理アクセスを利用します。
  • クエリを実行するユーザーから接続資格情報を非表示にします。
  • Spark データ ソース API を使用して外部データベースの読み取りと書き込みを行います。

JDBC Unity カタログ接続を使用するには、Spark オプションで databricks.connection を指定します。

df = (spark.read
  .format("jdbc")
  .option("databricks.connection", "<connection-name>")
  .option("query", "SELECT * FROM external_table")
  .load())

セットアップ手順については、 JDBC 接続を参照してください。

専用クラスターでカスタム コネクタを使用する

専用 (クラシック) クラスターでは、Databricks Runtime にバンドルされていないサードパーティの Spark データ ソース コネクタまたは JDBC ドライバーをインストールできます。

この方法は、次の場合に使用します。

  • MongoDB、Cassandra、Couchbase、Elasticsearch などのシステム用のサードパーティの Spark コネクタが必要です。
  • ランタイムにバンドルされていない特定のドライバー バージョンが必要です。
  • Unity カタログ接続を設定せずに、クラスターに JDBC ドライバーを直接インストールする必要があります。

コネクタまたはドライバーをインストールする

Compute>your cluster>Libraries>Install new を使用して、クラスターにライブラリをインストールします。 JAR をダウンロードまたはアップロードすることなく、Maven 座標を直接使用できます。 ライブラリを有効にするには、クラスターを再起動します。

データの読み取り

コネクタがインストールされたら、コネクタの形式名とその必要な接続オプションを使用してデータを読み取ります。

df = (spark.read
  .format("mongodb")
  .option("connection.uri", "mongodb://<hostname>:27017")
  .option("database", "<database-name>")
  .option("collection", "<collection-name>")
  .load())

データを書き込む

ソースにデータを書き戻すには、同じ形式名と接続オプションを使用します。

(df.write
  .format("mongodb")
  .mode("overwrite")
  .option("connection.uri", "mongodb://<hostname>:27017")
  .option("database", "<database-name>")
  .option("collection", "<collection-name>")
  .save())

考慮事項

専用クラスターでカスタム コネクタを使用する場合は、次の点に注意してください。

  • ドライバーまたはコネクタは、インストールされているクラスターでのみ使用できます。
  • カスタムサードパーティの Spark JAR は、Databricks SQL、サーバーレス、または標準アクセス モードのクラスターではサポートされていません。 これらのコンピューティングの種類では、バンドルされたコネクタまたは JDBC Unity カタログ接続を使用します。

PySpark カスタム データ ソース

Python DataSource API を使用すると、JAR または JVM ベースのライブラリを使用せずに、Pythonでカスタム データ コネクタを完全に構築できます。 これは、REST API、SaaS アプリケーション、または JDBC インターフェイスのないシステムに接続する必要がある場合、またはプログラムによって合成データを生成する場合に使用します。 この API では、バッチ読み取りとストリーミングの両方の読み取りと書き込みがサポートされています。

Note

PySpark カスタム データ ソースには、Databricks Runtime 15.4 LTS 以降が必要です。

セットアップ、例、API リファレンスについては、 PySpark カスタム データ ソースを参照してください。

統合戦略を比較する

次の表は、Spark Data Source API と Lakehouse Federation と Lakeflow Connect を比較して、ユース ケースに適したアプローチを選択するのに役立ちます。

特徴 Spark データ ソース API レイクハウスフェデレーション Lakeflow Connect
主なユース ケース 複雑な ETL、カスタム Spark ロジック、パススルー クエリ アドホック クエリ、BI レポート 大規模で自動化されたインジェスト
データの移動 Spark メモリに読み込まれる (エフェメラル) Spark メモリに読み込まれる (エフェメラル) Delta Lake にコピー済み（永続的）
クエリの実行 ネイティブ query オプションを使用した手動プッシュダウン Spark と SQL のフィルター、結合、集計の自動プッシュダウン 該当なし (テーブルの完全レプリケーション)
Governance Unity カタログ接続 (JDBC) またはシークレット スコープ Unity カタログ (フェデレーション カタログ) Unity Catalog（マネージド パイプライン）
最適な用途 Spark を最大限に柔軟に活用したい上級ユーザー向け ガバナンスを維持しながらデータ移動を最小限に抑える 運用 CDC とインジェスト パイプライン

サポートされているバンドル されたコネクタ

次のデータ ソースは Databricks Runtime にバンドルされており、Spark を介して直接呼び出すことができます。 読み取りと書き込みは、専用クラスターと標準クラスターでサポートされています。

Note

サーバーレス コンピューティングでの書き込みは、PostgreSQL、SQL Server、MySQL、Snowflake、Redshift でサポートされています。 サポートされている コネクタ オプションについては、バンドルされたコネクタのサーバーレス書き込み オプションを参照してください。

データ ソース spark.format() 名前
PostgreSQL "postgresql"
SQL Server "sqlserver"
MySQL と MariaDB "mysql"
Snowflake "snowflake"
Amazon Redshift "redshift"
Google ビッグクエリ（Google BigQuery） "bigquery"
Azure Synapse "SQLDW"
HTTP "http"

制限事項

Azure Databricksで Spark データ ソース API を使用する場合は、次の制限が適用されます。

  • バンドルされたデータ ソースの Spark オプションは、 query、 dbtable、およびコネクタ固有のオプションの小さなセットに限定されます。
  • カスタムサード パーティの Spark JAR は、専用クラスターにのみインストールできます。 サーバーレスクラスターまたは標準クラスターの場合は、バンドルされたコネクタまたは JDBC Unity カタログ接続を使用します。
  • PySpark カスタム データ ソースには、Databricks Runtime 15.4 LTS 以降が必要です。
  • Last updated on