Amazon
Web Services
Microsoft Azure
Google Cloud PlatformDatabricks SQL Connector for Python
Databricks SQL Connector for Python は、Python コードを使用して Databricks クラスターおよび Databricks SQLウェアハウスで SQL コマンドを実行できるようにする Python ライブラリです。Databricks SQL Connector for Python は、 pyodbc などの同様の Python ライブラリよりもセットアップと使用が簡単です。 このライブラリは PEP 249 – Python データベース API 仕様 v2.0 に準拠しています。
注
Databricks SQL Connector for Python には、Databricks 用の SQLAlchemy ダイアレクトも含まれています。 「 Databricks で SQLAlchemy を使用する」を参照してください。
要件
Python >=3.8 および <=3.11 を実行している開発マシン。
Databricks では、Python に含まれる venv によって提供されるものなど、Python 仮想環境を使用することをお勧めします。 仮想環境は、正しいバージョンの Python と Databricks SQL Connector for Python を一緒に使用するのに役立ちます。 仮想環境の設定と使用は、この記事の範囲外です。 詳細については、「 仮想環境の作成」を参照してください。
既存のクラスターまたはSQL ウェアハウス。
はじめに
pip install databricks-sql-connectorまたはpython -m pip install databricks-sql-connectorを実行して、開発用コンピューターに Databricks SQL Connector for Python ライブラリをインストールします。使用するクラスターまたは SQLウェアハウスに関する次の情報を収集します。
クラスターのサーバー・ホスト名。 これは、クラスターの [ Advanced Options] > [JDBC/ODBC ] タブの [Server ホスト名 ] の値から取得できます。
クラスターの HTTP パス。 これは、クラスターの [ Advanced Options] > [JDBC/ODBC ] タブの [HTTP Path] ( HTTP パス ) 値から取得できます。
認証
Databricks SQL Connector for Python では、次の Databricks 認証の種類がサポートされています。
注
Databricks のユーザー名とパスワードを使用した基本認証は、2024 年 7 月 10 日にサポートが終了しました。 Databricks 管理パスワードのサポート終了を参照してください。
Databricks個人用アクセストークン認証
個人用 アクセストークン認証 で Connector for を使用するにはDatabricks SQLPythonDatabricks まず、Databricks 個人用アクセストークンを作成する必要があります。これを行うには、「ワークスペース ユーザー向けの個人用アクセストークンDatabricks」のステップに従います。
Databricks SQL Connector for Python を認証するには、次のコード スニペットを使用します。 このスニペットは、次の環境変数が設定されていることを前提としています。
DATABRICKS_SERVER_HOSTNAMEをクラスターまたは SQLウェアハウスの [Server Hostname ] の値に設定します。DATABRICKS_HTTP_PATHで、クラスターまたは SQLウェアハウスの HTTP パス 値に設定します。DATABRICKS_TOKENを Databricks personal アクセストークンに設定します。
環境変数を設定するには、ご利用になっているオペレーティングシステムのドキュメントを参照してください。
from databricks import sql
import os
with sql.connect(server_hostname = os.getenv("DATABRICKS_SERVER_HOSTNAME"),
http_path = os.getenv("DATABRICKS_HTTP_PATH"),
access_token = os.getenv("DATABRICKS_TOKEN")) as connection:
# ...
OAuthマシン間（M2M）認証
Databricks SQL Connector for Python バージョン 2.5.0 以降では、OAuth マシン間 (M2M) 認証がサポートされています。 Databricks SDK for Pythonもインストールする必要があります (たとえばpip install databricks-sdkまたはpython -m pip install databricks-sdkを実行します)。
OAuth M2M 認証で Databricks SQL Connector for Python を使用するには、次の操作を行う必要があります。
Databricks ワークスペースに Databricks サービスプリンシパルを作成し、そのサービスプリンシパルの OAuth シークレットを作成します。
サービスプリンシパルとそのOAuth シークレットを作成するには、「 Databricksを使用してサービスプリンシパルで へのアクセスを認証するOAuth (OAuth M2M)」 を参照してください。サービスプリンシパルの UUID または Application ID の値と、サービスプリンシパルの シークレットの SecretOAuth 値をメモします。
そのサービスプリンシパルにクラスターまたはウェアハウスへのアクセス権を付与します。
サービスプリンシパルにクラスターまたはウェアハウスへのアクセスを許可するには、 「コンピュート権限」または「SQL ウェアハウスの管理」を参照してください。
Databricks SQL Connector for Python を認証するには、次のコード スニペットを使用します。 このスニペットは、次の環境変数が設定されていることを前提としています。
DATABRICKS_SERVER_HOSTNAMEをクラスターまたは SQLウェアハウスの [Server Hostname ] の値に設定します。DATABRICKS_HTTP_PATHで、クラスターまたは SQLウェアハウスの HTTP パス 値に設定します。DATABRICKS_CLIENT_IDは、サービスプリンシパルの UUID または アプリケーション ID の値に設定されます。DATABRICKS_CLIENT_SECRETで、サービスプリンシパルの OAuth シークレットの Secret 値に設定します。
環境変数を設定するには、ご利用になっているオペレーティングシステムのドキュメントを参照してください。
from databricks.sdk.core import Config, oauth_service_principal
from databricks import sql
import os
server_hostname = os.getenv("DATABRICKS_SERVER_HOSTNAME")
def credential_provider():
config = Config(
host = f"https://{server_hostname}",
client_id = os.getenv("DATABRICKS_CLIENT_ID"),
client_secret = os.getenv("DATABRICKS_CLIENT_SECRET"))
return oauth_service_principal(config)
with sql.connect(server_hostname = server_hostname,
http_path = os.getenv("DATABRICKS_HTTP_PATH"),
credentials_provider = credential_provider) as connection:
# ...
OAuthユーザー対マシン（U2M）認証
Databricks SQL Connector for Python バージョン 2.1.0 以上は 、OAuth user-to-machine (U2M) 認証をサポートしています。
OAuth U2M 認証を使用して Databricks SQL Connector for Python を認証するには、次のコード スニペットを使用します。 OAuth U2M 認証では、リアルタイムの人間のサインインと同意を使用して、ターゲット Databricks ユーザー アカウントを認証します。 このスニペットは、次の環境変数が設定されていることを前提としています。
DATABRICKS_SERVER_HOSTNAMEをクラスターまたは SQL ウェアハウスのサーバー ホスト名の値に設定します。DATABRICKS_HTTP_PATHをクラスターまたは SQL ウェアハウスのHTTP パス値に設定します。
環境変数を設定するには、ご利用になっているオペレーティングシステムのドキュメントを参照してください。
from databricks import sql
import os
with sql.connect(server_hostname = os.getenv("DATABRICKS_SERVER_HOSTNAME"),
http_path = os.getenv("DATABRICKS_HTTP_PATH"),
auth_type = "databricks-oauth") as connection:
# ...
例
次のコード例は、Databricks SQL コネクタ for Python を使用して、データのクエリと挿入、メタデータのクエリー、カーソルと接続の管理、およびログの構成を行う方法を示しています。
注
次のコード例は、認証に Databricks 個人用アクセストークンを使用する方法を示しています。 代わりに、他の使用可能な Databricks 認証の種類を使用するには、「 認証」を参照してください。
これらのコード例では、これらの環境変数から server_hostname、 http_path、および access_token 接続変数の値を取得します。
DATABRICKS_SERVER_HOSTNAMEこれは、要件の サーバ ホスト名 の値を表します。DATABRICKS_HTTP_PATHこれは、要件からの HTTP パス 値を表します。DATABRICKS_TOKENこれは、要件からのアクセストークンを表します。
これらの接続変数の値を取得するには、他の方法を使用できます。 環境変数の使用は、多くのアプローチの 1 つにすぎません。
データのクエリー
次のコード例は、Databricks SQL Connector for Python を呼び出して、クラスターまたは SQLウェアハウスで基本的な SQL コマンドを実行する方法を示しています。 このコマンドは、samples カタログのnyctaxiスキーマのtripsテーブルから最初の 2 行を返します。
from databricks import sql
import os
with sql.connect(server_hostname = os.getenv("DATABRICKS_SERVER_HOSTNAME"),
http_path = os.getenv("DATABRICKS_HTTP_PATH"),
access_token = os.getenv("DATABRICKS_TOKEN")) as connection:
with connection.cursor() as cursor:
cursor.execute("SELECT * FROM samples.nyctaxi.trips LIMIT 2")
result = cursor.fetchall()
for row in result:
print(row)
データの挿入
次の例は、少量のデータ (数千行) を挿入する方法を示しています。
from databricks import sql
import os
with sql.connect(server_hostname = os.getenv("DATABRICKS_SERVER_HOSTNAME"),
http_path = os.getenv("DATABRICKS_HTTP_PATH"),
access_token = os.getenv("DATABRICKS_TOKEN")) as connection:
with connection.cursor() as cursor:
cursor.execute("CREATE TABLE IF NOT EXISTS squares (x int, x_squared int)")
squares = [(i, i * i) for i in range(100)]
values = ",".join([f"({x}, {y})" for (x, y) in squares])
cursor.execute(f"INSERT INTO squares VALUES {values}")
cursor.execute("SELECT * FROM squares LIMIT 10")
result = cursor.fetchall()
for row in result:
print(row)
大量のデータの場合は、最初にデータをクラウドストレージにアップロードしてから、 COPY INTO コマンドを実行する必要があります。
クエリ メタデータ
メタデータを取得するための専用のメソッドがあります。 次の例では、サンプル テーブルの列に関するメタデータを取得します。
from databricks import sql
import os
with sql.connect(server_hostname = os.getenv("DATABRICKS_SERVER_HOSTNAME"),
http_path = os.getenv("DATABRICKS_HTTP_PATH"),
access_token = os.getenv("DATABRICKS_TOKEN")) as connection:
with connection.cursor() as cursor:
cursor.columns(schema_name="default", table_name="squares")
print(cursor.fetchall())
カーソルと接続を管理する
使用しなくなった接続とカーソルを閉じるのがベスト・プラクティスです。 これにより、Databricks クラスターと Databricks SQL ウェアハウスのリソースが解放されます。
コンテキストマネージャ(前の例で使用した with 構文)を使用してリソースを管理するか、明示的に closeを呼び出すことができます。
from databricks import sql
import os
connection = sql.connect(server_hostname = os.getenv("DATABRICKS_SERVER_HOSTNAME"),
http_path = os.getenv("DATABRICKS_HTTP_PATH"),
access_token = os.getenv("DATABRICKS_TOKEN"))
cursor = connection.cursor()
cursor.execute("SELECT * from range(10)")
print(cursor.fetchall())
cursor.close()
connection.close()
Unity Catalogボリューム内のファイルを管理する
Databricks SQL コネクタを使用すると、次の例に示すように、Unity Catalogボリュームにローカル ファイルを書き込んだり、ボリュームからファイルをダウンロードしたり、ボリュームからファイルを削除したりできます。
from databricks import sql
import os
# For writing local files to volumes and downloading files from volumes,
# you must set the staging_allows_local_path argument to the path to the
# local folder that contains the files to be written or downloaded.
# For deleting files in volumes, you must also specify the
# staging_allows_local_path argument, but its value is ignored,
# so in that case its value can be set for example to an empty string.
with sql.connect(server_hostname = os.getenv("DATABRICKS_SERVER_HOSTNAME"),
http_path = os.getenv("DATABRICKS_HTTP_PATH"),
access_token = os.getenv("DATABRICKS_TOKEN"),
staging_allowed_local_path = "/tmp/") as connection:
with connection.cursor() as cursor:
# Write a local file to the specified path in a volume.
# Specify OVERWRITE to overwrite any existing file in that path.
cursor.execute(
"PUT '/temp/my-data.csv' INTO '/Volumes/main/default/my-volume/my-data.csv' OVERWRITE"
)
# Download a file from the specified path in a volume.
cursor.execute(
"GET '/Volumes/main/default/my-volume/my-data.csv' TO '/tmp/my-downloaded-data.csv'"
)
# Delete a file from the specified path in a volume.
cursor.execute(
"REMOVE '/Volumes/main/default/my-volume/my-data.csv'"
)
ログの構成
Databricks SQL コネクタは、Python の 標準ログ モジュールを使用します。 ログ レベルは次のように構成できます。
from databricks import sql
import os, logging
logging.getLogger("databricks.sql").setLevel(logging.DEBUG)
logging.basicConfig(filename = "results.log",
level = logging.DEBUG)
connection = sql.connect(server_hostname = os.getenv("DATABRICKS_SERVER_HOSTNAME"),
http_path = os.getenv("DATABRICKS_HTTP_PATH"),
access_token = os.getenv("DATABRICKS_TOKEN"))
cursor = connection.cursor()
cursor.execute("SELECT * from range(10)")
result = cursor.fetchall()
for row in result:
logging.debug(row)
cursor.close()
connection.close()
テスティング
コードをテストするには、 pytestなどの Python テスト フレームワークを使用します。 Databricks REST API エンドポイントを呼び出さずに、または Databricks アカウントやワークスペースの状態を変更せずに、シミュレートされた条件下でコードをテストするには、 unittest.mockなどの Python モック ライブラリを使用できます。
たとえば、 Databricks個人用アクセストークンを使用してDatabricksワークスペースへの接続を返す get_connection_personal_access_token 関数と、接続を使用して samples カタログの nyctaxi スキーマの trips テーブルから指定された数のデータ行を取得する select_nyctaxi_trips 関数を含む helpers.py という名前の次のファイルがあるとします。
# helpers.py
from databricks import sql
from databricks.sql.client import Connection, List, Row, Cursor
def get_connection_personal_access_token(
server_hostname: str,
http_path: str,
access_token: str
) -> Connection:
return sql.connect(
server_hostname = server_hostname,
http_path = http_path,
access_token = access_token
)
def select_nyctaxi_trips(
connection: Connection,
num_rows: int
) -> List[Row]:
cursor: Cursor = connection.cursor()
cursor.execute(f"SELECT * FROM samples.nyctaxi.trips LIMIT {num_rows}")
result: List[Row] = cursor.fetchall()
return result
また、get_connection_personal_access_token 関数と select_nyctaxi_trips 関数を呼び出す main.py という名前の次のファイルがあるとします。
# main.py
from databricks.sql.client import Connection, List, Row
import os
from helpers import get_connection_personal_access_token, select_nyctaxi_trips
connection: Connection = get_connection_personal_access_token(
server_hostname = os.getenv("DATABRICKS_SERVER_HOSTNAME"),
http_path = os.getenv("DATABRICKS_HTTP_PATH"),
access_token = os.getenv("DATABRICKS_TOKEN")
)
rows: List[Row] = select_nyctaxi_trips(
connection = connection,
num_rows = 2
)
for row in rows:
print(row)
次の test_helpers.py という名前のファイルは、 select_nyctaxi_trips 関数が予期される応答を返すかどうかをテストします。 このテストは、ターゲットワークスペースへの実際の接続を作成するのではなく、 Connectionオブジェクトをモックします。 また、このテストでは、実際のデータに含まれるスキーマと値に準拠する一部のデータをモックします。 このテストでは、モックされた接続を介してモックされたデータが返され、モックされたデータ行の 1 つの値が期待値と一致するかどうかがチェックされます。
# test_helpers.py
import pytest
from databricks.sql.client import Connection, List, Row
from datetime import datetime
from helpers import select_nyctaxi_trips
from unittest.mock import create_autospec
@pytest.fixture
def mock_data() -> List[Row]:
return [
Row(
tpep_pickup_datetime = datetime(2016, 2, 14, 16, 52, 13),
tpep_dropoff_datetime = datetime(2016, 2, 14, 17, 16, 4),
trip_distance = 4.94,
fare_amount = 19.0,
pickup_zip = 10282,
dropoff_zip = 10171
),
Row(
tpep_pickup_datetime = datetime(2016, 2, 4, 18, 44, 19),
tpep_dropoff_datetime = datetime(2016, 2, 4, 18, 46),
trip_distance = 0.28,
fare_amount = 3.5,
pickup_zip = 10110,
dropoff_zip = 10110
)
]
def test_select_nyctaxi_trips(mock_data: List[Row]):
# Create a mock Connection.
mock_connection = create_autospec(Connection)
# Set the mock Connection's cursor().fetchall() to the mock data.
mock_connection.cursor().fetchall.return_value = mock_data
# Call the real function with the mock Connection.
response: List[Row] = select_nyctaxi_trips(
connection = mock_connection,
num_rows = 2)
# Check the value of one of the mocked data row's columns.
assert response[1].fare_amount == 3.5
select_nyctaxi_trips 関数には SELECT ステートメントが含まれているため、trips テーブルの状態は変更されないため、この例ではモック作成は絶対に必要というわけではありません。ただし、モックを使用すると、ワークスペースとの実際の接続が確立されるのを待たずに、テストをすばやく実行できます。 また、モックを使用すると、 INSERT INTO 、 UPDATE 、 DELETE FROMなど、テーブルの状態を変更する可能性のある関数に対して、シミュレートされたテストを複数回実行できます。
API リファレンス
パッケージ
databricks-sql-connector
使い： pip install databricks-sql-connector
Python パッケージインデックス (PyPI) の databricks-sql-connector も参照してください。
クラス
選択されたクラスには、次のものが含まれます。
クラス |
|---|
Databricksコンピュートリソースに関するセッション。 |
データ・レコードをトラバースするためのメカニズム。 |
SQL クエリ結果内のデータの行。 |
Connection クラス
Connectionオブジェクトを作成するには、次の引数を使用してdatabricks.sql.connectメソッドを呼び出します。
パラメーター |
|---|
種類： クラスターまたは SQLウェアハウスのサーバーのホスト名。 サーバーのホスト名を取得するには、この記事で前述した手順を参照してください。 このパラメーターは必須です。 例： |
種類： クラスターまたは SQLウェアハウスの HTTP パス。 HTTP パスを取得するには、この記事で前述した手順を参照してください。 このパラメーターは必須です。 例: クラスターの |
種類： Databricks の認証設定に関する情報。 詳細については、「 認証」を参照してください。 |
種類： Spark セッション構成パラメーターのディクショナリ。 構成の設定は、 デフォルトは このパラメーターはオプションです。 例： |
種類： クライアントが行うすべての RPC 要求の HTTP ヘッダーに設定する追加の (キーと値のペア。 一般的な使用法では、追加の HTTP ヘッダーは設定されません。 デフォルトは このパラメーターはオプションです。 バージョン 2.0 以降 |
種類： 接続に使用する初期カタログ。 デフォルトから このパラメーターはオプションです。 バージョン 2.0 以降 |
種類： 接続に使用する初期スキーマ。 デフォルトから このパラメーターはオプションです。 バージョン 2.0 以降 |
種類：
バージョン 2.8 以降 |
選択された Connection 方法には、次のものがあります。
メソッド |
|---|
データベースへの接続を閉じ、サーバー上のすべての関連リソースを解放します。 この接続への追加の呼び出しは、 パラメーターはありません。 戻り値はありません。 |
データベース内のレコードのトラバーサルを可能にする新しい パラメーターはありません。 |
Cursor クラス
Cursor オブジェクトを作成するには、Connection クラスの cursor メソッドを呼び出します。
選択された Cursor 属性には、次のものがあります。
属性 |
|---|
読み取り/書き込みアクセス。 |
各 7 項目 読み取り専用アクセス。 |
選択された Cursor 方法には、次のものがあります。
メソッド |
|---|
カーソルが開始したデータベース クエリまたはコマンドの実行を中断します。 サーバー上の関連リソースを解放するには、 パラメーターはありません。 戻り値はありません。 |
カーソルを閉じ、サーバー上の関連リソースを解放します。 既に閉じているカーソルを閉じると、エラーがスローされる可能性があります。 パラメーターはありません。 戻り値はありません。 |
データベースを準備し、クエリーまたはコマンドを実行します。 戻り値はありません。 パラメーター:
種類： 準備して実行するクエリーまたはコマンド。 このパラメーターは必須です。
cursor.execute(
'SELECT * FROM samples.nyctaxi.trips WHERE pickup_zip="10019" LIMIT 2'
)
cursor.execute(
'SELECT * FROM samples.nyctaxi.trips WHERE zip=%(pickup_zip)s LIMIT 2',
{ 'pickup_zip': '10019' }
)
原稿種別: 辞書
このパラメーターはオプションです。 デフォルトは |
データベースを準備し、 戻り値はありません。 パラメーター:
種類： 準備して実行するクエリーまたはコマンド。 このパラメーターは必須です。
タイプ:
このパラメーターは必須です。 |
カタログに関するメタデータ クエリを実行します。 実際の結果は、 結果セットの重要なフィールドには、次のものがあります。
パラメーターはありません。 戻り値はありません。 バージョン 1.0 以降 |
スキーマに関するメタデータ クエリを実行します。 実際の結果は、 結果セットの重要なフィールドには、次のものがあります。
戻り値はありません。 バージョン 1.0 以降 パラメーター:
種類： 情報を取得するカタログ名。 このパラメーターはオプションです。
種類： 情報を取得するスキーマ名。 このパラメーターはオプションです。 |
テーブルとビューに関するメタデータ クエリを実行します。 実際の結果は、 結果セットの重要なフィールドには、次のものがあります。
戻り値はありません。 バージョン 1.0 以降 パラメーター
種類： 情報を取得するカタログ名。 このパラメーターはオプションです。
種類： 情報を取得するスキーマ名。 このパラメーターはオプションです。
種類： 情報を取得するテーブル名。 このパラメーターはオプションです。
種類： 照合するテーブルタイプのリスト ( このパラメーターはオプションです。 |
列に関するメタデータ クエリを実行します。 実際の結果は、 結果セットの重要なフィールドには、次のものがあります。
戻り値はありません。 バージョン 1.0 以降 パラメーター:
種類： 情報を取得するカタログ名。 このパラメーターはオプションです。
種類： 情報を取得するスキーマ名。 このパラメーターはオプションです。
種類： 情報を取得するテーブル名。 このパラメーターはオプションです。
種類： 情報を取得する列名。 このパラメーターはオプションです。 |
クエリーのすべて (または残りのすべて) 行を取得します。 パラメーターはありません。 クエリーのすべての (または残りのすべての) 行を、
|
クエリーの次の行を取得します。 クエリの次の行の最大 フェッチする行数が
パラメーター:
種類： 取得する次の行の数。 このパラメーターはオプションです。 指定しない場合は、 例： |
データセットの次の行を取得します。 パラメーターはありません。 データセットの次の行を Python
|
クエリーのすべて (または残りのすべて) 行を PyArrow パラメーターはありません。 クエリーのすべて (または残りのすべて) 行を PyArrow テーブルとして返します。
バージョン 2.0 以降 |
クエリーの次の行を PyArrow クエリの次の行の
バージョン 2.0 以降 パラメーター:
種類： 取得する次の行の数。 このパラメーターはオプションです。 指定しない場合は、 例： |
Row クラス
行クラスは、個々の結果行を表すタプルのようなデータ構造です。 行に "my_column"という名前の列が含まれている場合は、 row.my_columnを介して row の "my_column" フィールドにアクセスできます。数値インデックスを使用して、 row[0]などのフィールドにアクセスすることもできます。 列名が属性メソッド名として許可されていない場合 (たとえば、数字で始まる場合)、フィールドには row["1_my_column"]としてアクセスできます。
バージョン 1.0 以降
選択された Row 方法は次のとおりです。
フィールド名でインデックス付けされた行の辞書表現を返します。 重複するフィールド名がある場合、重複するフィールドの 1 つ (ただし、1 つだけ) が辞書に返されます。 どの重複フィールドが返されるかは定義されていません。 パラメーターはありません。 フィールドの |
型変換
次の表は、Apache Spark SQL データ型を同等の Python データ型にマップします。
Apache Spark SQL データ型 |
Python データ型 |
|---|---|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
トラブルシューティング
tokenAuthWrapperInvalidAccessToken: Invalid access token メッセージ
問題: コードを実行すると、 Error during request to server: tokenAuthWrapperInvalidAccessToken: Invalid access tokenのようなメッセージが表示されます。
考えられる原因: access_token に渡された値が有効な Databricks 個人用アクセストークンではありません。
推奨される修正: access_token に渡された値が正しいことを確認し、再試行してください。
gaierror(8, 'nodename nor servname provided, or not known') メッセージ
問題: コードを実行すると、 Error during request to server: gaierror(8, 'nodename nor servname provided, or not known')のようなメッセージが表示されます。
考えられる原因: server_hostname に渡された値が正しくホスト名ではありません。
推奨される修正: server_hostname に渡された値が正しいことを確認し、再試行してください。
サーバーのホスト名の検索の詳細については、「 Databricks コンピュート リソースの接続の詳細を取得する」を参照してください。