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のUnity Catalog接続を使います

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

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

  • JDBC をサポートする任意のデータベースに対して、独自の JDBC ドライバー JAR を使用します。
  • 一度接続を作成し、サーバーレス、標準、専用コンピュートで再利用します。
  • 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 レイクハウスフェデレーション レイクフロー・コネクト
主なユース ケース 複雑な 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"

Databricks Runtime 20 向けの BigQuery JDBC ワークロードの移行

Databricks Runtime 20からは、バンドルされているBigQuery JDBCドライバーは含まれていません。 Databricks Runtime 19以前でもドライバーは引き続き含まれています。 この変更は、 jdbc:bigquery: URLでSpark JDBCを使用するクラシックなコンピュートワークロードに影響します。

影響を受けるワークロードを見つけるには、コードと設定で jdbc:bigquery: URLを検索してください。 このJDBCパスは、 .format("bigquery") を使い、引き続き利用可能なバンドルのGoogle BigQueryコネクタとは別です。 BigQuery JDBCドライバーがインストールされていないと、影響を受けたワークロードはドライバーが見つからないエラーで失敗します。

影響を受けるワークロードをDatabricks Runtime 20以降に移行する前に、以下のいずれかのオプションを選択してください。

Googleで利用可能なドライバーについては、 BigQuery ODBCおよびJDBCドライバーを参照してください。

制限事項

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

  • バンドルされたデータ ソースの Spark オプションは、 query、 dbtable、およびコネクタ固有のオプションの小さなセットに限定されます。
  • カスタムサードパーティ製Spark JARは専用のコンピュートにのみインストール可能です。 サーバーレスまたは標準的なコンピュートの場合は、バンドルコネクターやJDBC Unity Catalog接続をご利用ください。
  • PySpark カスタム データ ソースには、Databricks Runtime 15.4 LTS 以降が必要です。
  • 最終更新: