Python 用 Databricks SQL コネクタ

Databricks SQL Connector for Python は Python ライブラリであり、Python コードを使用して、Azure Databricks の汎用コンピューティングおよび Databricks SQL ウェアハウスで SQL コマンドを実行できます。 Databricks SQL Connector for Python は、pyodbc のような類似の Python ライブラリよりも設定と使用が簡単です。このライブラリは、PEP 249 – Python Database API Specification v2.0 に準拠しています。

重要

Databricks SQL Connector for Python バージョン 3.0.0 以降では、ネイティブのパラメーター化されたクエリ実行がサポートされています。これにより、SQL インジェクションが防止され、クエリのパフォーマンスが向上します。以前のバージョンでは、インラインパラメーター化された実行が使用されていたため、SQL インジェクションから安全ではなく、その他の欠点があります。詳細については、「ネイティブパラメーターの使用」を参照してください。

Python 用 Databricks SQL コネクタでは、Azure Databricks の SQLAlchemy 方言もサポートされていますが、これらの機能を使用するにはインストールする必要があります。「Azure Databricks で SQLAlchemy を使用する」を参照してください。

要件

Python 3.8 以降を実行している開発マシン。
Databricks では、Python に含まれる venv によって提供されるもののような Python 仮想環境を使用することをお勧めします。仮想環境を使用すると、正しいバージョンの Python と Databricks SQL Connector for Python の組み合わせが確実に使用されるようになります。仮想環境のセットアップと使用は、この記事の範囲外です。詳細については、「仮想環境を構築する」を参照してください。
既存の汎用コンピューティングまたは SQL ウェアハウス。

概要

Python 用 Databricks SQL コネクタをインストールします。 PyArrow は Databricks SQL Connector for Python のオプションの依存関係であり、バージョン 4.0.0 以降のコネクタでは既定ではインストールされません。 PyArrow がインストールされていない場合、CloudFetch やその他の Apache Arrow 機能などの機能は使用できません。これにより、大量のデータのパフォーマンスに影響する可能性があります。
- リーンコネクタをインストールするには、次のコマンドを使用します。
```
pip install databricks-sql-connector
```
- PyArrow を含む完全なコネクタをインストールするには、次のコマンドを使用します。
```
pip install databricks-sql-connector[pyarrow]
```
使用する汎用コンピューティングまたは SQL ウェアハウスについて、次の情報を収集します。

汎用コンピューティング
- 汎用コンピューティングのサーバーホスト名。これは、汎用コンピューティングの [] タブの >] の値から取得できます。
- 汎用コンピューティングの HTTP パス。これは、汎用コンピューティングの [] タブの >] の値から取得できます。
注

SQL コネクタは、ジョブコンピューティングへの接続をサポートしていません。

SQL ウェアハウス
- SQL ウェアハウスのサーバーホスト名。これは、SQL ウェアハウスの [接続の詳細] タブにある [サーバーホスト名] の値から取得できます。
- SQL ウェアハウスの HTTP パス。これは、SQL ウェアハウスの [接続の詳細] タブにある [HTTP パス] の値から取得できます。

認証

Databricks SQL Connector for Python では、次の種類の Azure Databricks 認証がサポートされています。

Databricks 個人用アクセストークン認証
Microsoft Entra ID トークン認証
OAuth マシン間 (M2M) 認証
OAuth ユーザーマシン間 (U2M) 認証

Databricks SQL Connector for Python では、次の種類の Azure Databricks 認証はまだサポートされていません。

Databricks 個人用アクセストークン認証

Databricks SQL Connector for Python と Azure Databricks 個人用アクセストークン認証を使用するには、最初に Azure Databricks 個人用アクセストークンを作成する必要があります。これを行うには、「ワークスペースユーザーの個人用アクセストークンを作成する」の手順に従います。

Databricks SQL Connector for Python を認証するには、次のコードスニペットを使用します。このスニペットでは、次の環境変数が設定されていることを想定しています。

DATABRICKS_SERVER_HOSTNAMEは、万能コンピューティングまたは SQL ウェアハウスの サーバーホスト名 の値に設定されます。
DATABRICKS_HTTP_PATH: All-Purpose Compute または SQL ウェアハウスの HTTP パス値に設定。
DATABRICKS_TOKEN: Azure Databricks 個人用アクセストークンに設定。

環境変数を設定するには、オペレーティングシステムのドキュメントを参照してください。

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.7.0 以降では、OAuth マシン間 (M2M) 認証がサポートされています。 Databricks SDK for Python 0.18.0 以降も (pip install databricks-sdk や python -m pip install databricks-sdk を実行するなどをして) インストールする必要があります。

Databricks SQL Connector for Python と OAuth M2M 認証を使用するには、次を実行する必要があります。

Azure Databricks ワークスペースで Azure Databricks サービスプリンシパルを作成し、そのサービスプリンシパル用の OAuth シークレットを作成します。

サービスプリンシパルとその OAuth シークレットを作成するには、「 OAuth を使用して Azure Databricks へのサービスプリンシパルアクセスを承認する」を参照してください。サービスプリンシパルの UUID または アプリケーション ID の値と、サービスプリンシパルの OAuth シークレットのシークレット 値をメモしておきます。
そのサービスプリンシパルに、All-Purpose Compute またはウェアハウスへのアクセス権を付与します。

サービスプリンシパルに万能コンピューティングまたはウェアハウスへのアクセス権を付与するには、「コンピューティングのアクセス許可」または「SQL ウェアハウスの管理」を参照してください。

DATABRICKS_SERVER_HOSTNAME は、万能コンピューティングまたは SQL ウェアハウスの サーバーホスト名 の値に設定されます。
DATABRICKS_HTTP_PATH: All-Purpose Compute または SQL ウェアハウスの HTTP パス値に設定。
DATABRICKS_CLIENT_IDは、サービスプリンシパルの UUID または アプリケーション ID の値に設定されます。
DATABRICKS_CLIENT_SECRET: サービスプリンシパルの OAuth シークレットのシークレット値に設定。

環境変数を設定するには、オペレーティングシステムのドキュメントを参照してください。

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:
# ...

Microsoft Entra ID トークン認証

Databricks SQL Connector for Python と Microsoft Entra ID トークン認証を使用するには、Databricks SQL Connector for Python に Microsoft Entra ID トークンを提供する必要があります。 Microsoft Entra ID アクセストークンを作成するには、次を実行します。

Azure Databricks ユーザーの場合は、Azure CLI を使用できます。「Microsoft Entra ID トークンを手動で取得する」を参照してください。
Microsoft Entra ID サービスプリンシパルについては、「サービスプリンシパルのトークンを取得する」を参照してください。 Microsoft Entra ID マネージドサービスプリンシパルを作成するには、「サービスプリンシパル」を参照してください。

Microsoft Entra ID トークンの既定の有効期間は約 1 時間です。新しい Microsoft Entra ID トークンを作成するには、このプロセスを繰り返します。

DATABRICKS_SERVER_HOSTNAMEを、汎用コンピューティングまたは SQL ウェアハウスのサーバーホスト名の値に設定します。
DATABRICKS_HTTP_PATHを汎用コンピューティングまたは SQL ウェアハウスの HTTP パス値に設定します。
DATABRICKS_TOKEN を、Microsoft Entra ID トークンに設定します。

環境変数を設定するには、オペレーティングシステムのドキュメントを参照してください。

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 ユーザーマシン間 (U2M) 認証

Databricks SQL Connector for Python バージョン 2.7.0 以降では、OAuth ユーザーマシン間 (U2M) 認証がサポートされています。 Databricks SDK for Python 0.19.0 以降もインストールする必要があります (たとえば、pip install databricks-sdk または python -m pip install databricks-sdk を実行します)。

OAuth U2M 認証を使用して Databricks SQL Connector for Python を認証するには、次のコードスニペットを使用します。 OAuth U2M 認証では、ユーザーのリアルタイムのサインインと同意を使用して、ターゲットの Azure 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 Connector for Python を使用して、データのクエリと挿入、メタデータのクエリ、カーソルと接続の管理、Unity カタログでのファイルの管理、ログ記録の構成を行う方法を示しています。

注

以下のコード例では、認証に Azure Databricks 個人用アクセストークンを使用する方法を示します。別の認証の種類を使用するには、「認証」を参照してください。

このコード例では、これらの環境変数から server_hostname、http_path、access_token 接続変数の値を取得します。

DATABRICKS_SERVER_HOSTNAME は、要件からのサーバーホスト名を表します。
DATABRICKS_HTTP_PATH は、要件からの HTTP パスを表します。
DATABRICKS_TOKEN は、要件からのアクセストークンを表します。

User-Agent の設定

次のコード例では、使用状況の追跡用に User-Agent アプリケーションの product_name を設定する方法を示します。

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"),
                 user_agent_entry = "product_name") as connection:
  with connection.cursor() as cursor:
    cursor.execute("SELECT 1 + 1")
    result = cursor.fetchall()

    for row in result:
      print(row)

クエリデータ

次のコード例は、Databricks SQL Connector for Python を呼び出して、汎用コンピューティングまたは SQL ウェアハウスで基本的な SQL コマンドを実行する方法を示しています。このコマンドは、trips カタログのsamples スキーマのnyctaxi テーブルから最初の 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)

クエリタグ

重要

この機能は、プライベートプレビューにあります。アクセス権を要求するには、アカウントチームにお問い合わせください。

次の例は、追跡と分析のために SQL クエリにキー値タグをアタッチする方法を示しています。 system.query.history テーブルにクエリタグが表示されます。

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"),
    session_configuration = {
        'query_tags': 'team:engineering,dashboard:abc123,env:prod'
    }
) as connection:
    with connection.cursor() as cursor:
        cursor.execute("SELECT * FROM samples.nyctaxi.trips LIMIT ?", [2])
        result = cursor.fetchall()
        # Query is now tagged and trackable in system.query.history
        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)]

    cursor.executemany("INSERT INTO squares VALUES (?, ?)", squares)

    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())

カーソルと接続を管理する

使用されなくなった接続とカーソルを閉じるのがベストプラクティスです。これにより、Azure 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 Connector を使用すると、次の例に示すように、Unity Catalog ボリュームにローカルファイルを書き込み、ボリュームからファイルをダウンロードし、ボリュームからファイルを削除することができます。

from databricks import sql
import os

# For writing local files to volumes and downloading files from volumes,
# you must set the staging_allowed_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_allowed_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 '/tmp/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 テストフレームワークを使用します。 Azure Databricks REST API エンドポイントを呼び出したり、Azure Databricks アカウントまたはワークスペースの状態を変更したりせずに、シミュレートされた条件下でコードをテストするには、unittest.mock などの Python モックライブラリを使用できます。

たとえば、Azure Databricks 個人用アクセストークンを使用して Azure Databricks ワークスペースへの接続を返すhelpers.py関数を含む get_connection_personal_access_token という名前の次のファイルと、接続を使用してselect_nyctaxi_trips カタログのtrips スキーマのsamples テーブルから指定された数のデータ行を取得するnyctaxi関数を指定します。

# 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("SELECT * FROM samples.nyctaxi.trips LIMIT ?", [num_rows])
  result: List[Row] = cursor.fetchall()
  return result

また、main.py と get_connection_personal_access_token 関数を呼び出す select_nyctaxi_trips という名前の次のようなファイルがあるとします。

# 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 オブジェクトをモックします。また、このテストでは、実際のデータでのスキーマと値と一致するデータをモックします。テストはモック接続を介してモックデータを返し、モックデータ行の値のいずれかが期待値と一致するかどうかを確認します。

# 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 パッケージの API リファレンスが含まれています。 Python パッケージインデックス (PyPI) の databricks-sql-connector を参照してください。

モジュール

databricks.sql パッケージの databricks-sql-connector モジュールには、SQL ウェアハウスへの接続を初期化するメソッドが含まれています。

connect メソッド

SQL ウェアハウスへの接続を初期化します。 Connection オブジェクトを返します。

パラメーター	タイプ	Description
`server_hostname`	`str`	必須。 `adb-1234567890123456.7.azuredatabricks.net`など、汎用コンピューティングまたは SQL ウェアハウスのサーバーホスト名。サーバーのホスト名を取得するには、「はじめに」の手順を参照してください。
`http_path`	`str`	必須。汎用コンピューティングまたは SQL ウェアハウスの HTTP パス。たとえば、汎用コンピューティングの場合は `sql/protocolv1/o/1234567890123456/1234-567890-test123` 、SQL ウェアハウスの場合は `/sql/1.0/warehouses/a1b234c567d8e9fa` 。 HTTP パスを取得するには、「はじめに」の手順を参照してください。
`access_token`、 `auth_type`、 `credentials_provider`、 `password`、 `username`	`str`	Azure Databricks の認証設定に関する情報。詳細については、「認証」を参照してください。
`session_configuration`	`dict[str, Any]`	Spark セッション構成パラメーターの辞書。構成の設定は、`SET key=val` SQL コマンドを使用することと同じです。 SQL コマンド `SET -v` を実行して、使用可能な構成の完全なリストを取得します。既定値は `None` です。例: `{"spark.sql.variable.substitute": True}`
`http_headers`	`List[Tuple[str, str]]]`	Optional. クライアントが行う RPC 要求ごとに HTTP ヘッダーで設定する追加の (キー、値) ペア。一般的な使用法では、追加の HTTP ヘッダーは設定されません。既定値は `None` です。
`catalog`	`str`	Optional. 接続に使用する初期カタログです。既定値は `None` です (この場合は既定のカタログが使用されます。通常は `hive_metastore` が使用されます)。
`schema`	`str`	Optional. 接続に使用する初期スキーマです。既定値は `None` です (この場合、既定のスキーマ (通常は `default`) が使用されます)。バージョン 2.0 以降
`use_cloud_fetch`	`bool`	Optional. データのチャンクをダウンロードするために、フェッチ要求をクラウドオブジェクトストアに直接送信するかどうか。既定値は `True` です。フェッチ要求を Azure Databricks に直接送信するには、 `False` に設定します。 `use_cloud_fetch` が `True` に設定されていても、ネットワークアクセスがブロックされている場合、フェッチ要求は失敗します。バージョン 2.8 以降
`user_agent_entry`	`str`	Optional. 使用状況の追跡のために HTTP 要求ヘッダーに含める User-Agent エントリ。既定値は `PyDatabricksSqlConnector` です。

`Connection` クラス

コンピューティングまたは SQL ウェアハウスへの接続を表します。

メソッド

Connection クラスには、次のメソッドがあります。

メソッド	Description
`close`	データベースへの接続を閉じ、サーバー上のすべての関連リソースを解放します。この接続への呼び出しを追加すると、`Error` がスローされます。パラメーターはありません。戻り値はありません。
`cursor`	データベース内のレコードのトラバーサルを有効にする新しい Cursor オブジェクトを返します。パラメーターはありません。

`Cursor` クラス

データレコードを走査する仕組みを示します。

Cursor オブジェクトを作成するには、cursorのメソッドを呼び出します。

属性

次のような Cursor 属性があります。

特性 Description

arraysize fetchmany メソッドで使用され、内部バッファーサイズを指定します。これは、実際にサーバーから一度にフェッチされる行数でもあります。既定値は 10000 です。狭い結果 (各行に多くのデータが含まれていない結果) の場合は、パフォーマンスを向上させるために、この値を大きくする必要があります。読み取り/書き込みアクセス。

特性	Description
`arraysize`	`fetchmany` メソッドで使用され、内部バッファーサイズを指定します。これは、実際にサーバーから一度にフェッチされる行数でもあります。既定値は `10000` です。狭い結果 (各行に多くのデータが含まれていない結果) の場合は、パフォーマンスを向上させるために、この値を大きくする必要があります。読み取り/書き込みアクセス。
`description`	Python の `list` オブジェクトの `tuple` が含まれます。これらの `tuple` オブジェクトにはそれぞれ 7 つの値が含まれ、各 `tuple` オブジェクトの最初の 2 つの項目に、次のように単一の結果列を説明する情報が含まれます。 `name`: 列の名前。 `type_code`: 列の型を表す文字列。たとえば、整数列の型コードは `int` になります。各 7 項目の `tuple` オブジェクトの残りの 5 項目は実装されておらず、それらの値は定義されていません。通常、4 `None` 値の後に 1 つの `True` 値が返されます。読み取り専用アクセス。

description

Python の list オブジェクトの tuple が含まれます。これらの tuple オブジェクトにはそれぞれ 7 つの値が含まれ、各 tuple オブジェクトの最初の 2 つの項目に、次のように単一の結果列を説明する情報が含まれます。

name: 列の名前。
type_code: 列の型を表す文字列。たとえば、整数列の型コードは int になります。各 7 項目の tuple オブジェクトの残りの 5 項目は実装されておらず、それらの値は定義されていません。通常、4 None 値の後に 1 つの True 値が返されます。読み取り専用アクセス。

メソッド

次のような Cursor メソッドがあります。

メソッド	Description
`cancel`	カーソルが起動したデータベースクエリまたはコマンドの実行を中断します。サーバー上の関連付けられているリソースを解放するには、`close` メソッドを呼び出した後、`cancel` メソッドを呼び出します。パラメーターはありません。戻り値はありません。
`close`	カーソルを閉じ、サーバー上の関連リソースを解放します。既に閉じているカーソルを閉じると、エラーがスローされる可能性があります。パラメーターはありません。戻り値はありません。
`execute`	データベースクエリまたはコマンドを準備して実行します。パラメーター: `operation`：必須。準備してから実行するクエリまたはコマンド。型: `strparameters` パラメーターを指定しない例: `cursor.execute('SELECT * FROM samples.nyctaxi.trips LIMIT 2')` `parameters` パラメーターの例 (ネイティブ位置指定パラメーターを使用):`cursor.execute('SELECT * FROM samples.nyctaxi.trips WHERE pickup_zip = ? LIMIT ?', ['10019', 2])` `parameters`: 省略可能。 `operation` パラメーターと一緒に使用するパラメーターのシーケンス。既定では、 `None`です。型: `dictionary` 戻り値はありません。
`executemany`	`seq_of_parameters` 引数内のすべてのパラメーターシーケンスを使用して、データベースクエリまたはコマンドを準備して実行します。最終的な結果セットだけが保持されます。パラメーター: `operation`：必須。準備してから実行するクエリまたはコマンド。型: `str` `seq_of_parameters`：必須。 `operation` パラメーターで使用するパラメーター値の多数のセットのシーケンス。型: `list` の `dict` 戻り値はありません。
`catalogs`	カタログに関するメタデータクエリを実行します。実際の結果は、`fetchmany` または `fetchall` を使用してフェッチする必要があります。結果セットの重要なフィールドには、以下が含まれます。フィールド名: `TABLE_CAT`。カタログの名前です。型: `str` パラメーターはありません。戻り値はありません。バージョン 1.0 以降
`schemas`	スキーマに関するメタデータクエリを実行します。実際の結果は、`fetchmany` または `fetchall` を使用してフェッチする必要があります。結果セットの重要なフィールドには、以下が含まれます。フィールド名: `TABLE_SCHEM`。スキーマの名前です。型: `str` フィールド名: `TABLE_CATALOG`。スキーマが属するカタログ。型: `str` パラメーター: `catalog_name`: 省略可能。情報を取得するカタログ名。 `%` 文字は、ワイルドカードとして解釈されます。型: `str` `schema_name`: 省略可能。情報を取得するスキーマ名。 `%` 文字は、ワイルドカードとして解釈されます。型: `str` 戻り値はありません。バージョン 1.0 以降
`tables`	テーブルとビューに関するメタデータクエリを実行します。実際の結果は、`fetchmany` または `fetchall` を使用してフェッチする必要があります。結果セットの重要なフィールドには、以下が含まれます。フィールド名: `TABLE_CAT`。テーブルが属するカタログ。型: `str` フィールド名: `TABLE_SCHEM`。テーブルが属するスキーマ。型: `str` フィールド名: `TABLE_NAME`。テーブルの名前です。型: `str` フィールド名: `TABLE_TYPE`。関係の種類。たとえば `VIEW` や `TABLE` (Databricks Runtime 10.4 LTS 以降および Databricks SQL に適用されます。以前のバージョンの Databricks Runtime では空の文字列が返されます)。型: `str` パラメーター: `catalog_name`: 省略可能。情報を取得するカタログ名。 `%` 文字は、ワイルドカードとして解釈されます。型: `str` `schema_name`: 省略可能。情報を取得するスキーマ名。 `%` 文字は、ワイルドカードとして解釈されます。型: `str` `table_name`: 省略可能。情報を取得するテーブル名。 `%` 文字は、ワイルドカードとして解釈されます。型: `str` `table_types`: 省略可能。照合するテーブル型のリスト (例:`TABLE`、`VIEW`)。型: `List[str]` 戻り値はありません。バージョン 1.0 以降
`columns`	列に関するメタデータクエリを実行します。実際の結果は、`fetchmany` または `fetchall` を使用してフェッチする必要があります。結果セットの重要なフィールドには、以下が含まれます。フィールド名: `TABLE_CAT`。列が属するカタログ。型: `str` フィールド名: `TABLE_SCHEM`。列が属するスキーマ。型: `str` フィールド名: `TABLE_NAME`。列が属するテーブルの名前。型: `str` フィールド名: `COLUMN_NAME`。列の名前。型: `str` パラメーター: `catalog_name`: 省略可能。情報を取得するカタログ名。 `%` 文字は、ワイルドカードとして解釈されます。型: `str` `schema_name`: 省略可能。情報を取得するスキーマ名。 `%` 文字は、ワイルドカードとして解釈されます。型: `str` `table_name`: 省略可能。情報を取得するテーブル名。 `%` 文字は、ワイルドカードとして解釈されます。型: `str` `column_name`: 省略可能。情報を取得する列名。 `%` 文字は、ワイルドカードとして解釈されます。型: `str` 戻り値はありません。バージョン 1.0 以降
`fetchall`	クエリのすべて (または残りすべて) の行を取得します。パラメーターはありません。クエリのすべての (または残りの) 行を、`list` オブジェクトの Python `Row`として返します。 `Error` メソッドの前回の呼び出しでデータが返されなかった場合、または `execute` の呼び出しがまだ行われていない場合は、`execute` をスローします。
`fetchmany`	クエリの次の行を取得します。パラメーター: `size`: 省略可能。取得する次の行の数。指定しない場合、`arraysize` 属性の値が使用されます。次のコマンドを入力します: `int` 例: `cursor.fetchmany(10)` クエリの次の最大 `size` (または `arraysize` が指定されていない場合は `size` 属性) 個の行を、Python の `list` オブジェクトの `Row` として返します。残りのフェッチする行が `size` より少ない場合は、残りの行がすべて返されます。 `Error` メソッドの前回の呼び出しでデータが返されなかった場合、または `execute` の呼び出しがまだ行われていない場合は、`execute` をスローします。
`fetchone`	データセットの次の行を取得します。パラメーターはありません。データセットの次の行を Python `tuple` オブジェクトとして 1 つのシーケンスとして返すか、使用可能なデータがなくなった場合は `None` を返します。 `Error` メソッドの前回の呼び出しでデータが返されなかった場合、または `execute` の呼び出しがまだ行われていない場合は、`execute` をスローします。
`fetchall_arrow`	クエリのすべて (または残りすべて) の行を、PyArrow `Table` オブジェクトとして取得します。非常に大量のデータを返すクエリは、メモリ消費を減らす代わりに、`fetchmany_arrow` を使用する必要があります。パラメーターはありません。クエリのすべて (または残りすべて) の行を PyArrow テーブルとして返します。 `Error` メソッドの前回の呼び出しでデータが返されなかった場合、または `execute` の呼び出しがまだ行われていない場合は、`execute` をスローします。バージョン 2.0 以降
`fetchmany_arrow`	クエリの次の行を PyArrow `Table` オブジェクトとして取得します。パラメーター: `size`: 省略可能。取得する次の行の数。指定しない場合、`arraysize` 属性の値が使用されます。次のコマンドを入力します: `int` 例: `cursor.fetchmany_arrow(10)` Python PyArrow `size` オブジェクトとしてクエリの次の行の`arraysize`引数 (または、`size`が指定されていない場合は`Table`属性) までを返します。 `Error` メソッドの前回の呼び出しでデータが返されなかった場合、または `execute` の呼び出しがまだ行われていない場合は、`execute` をスローします。バージョン 2.0 以降

`Row` クラス

行クラスは、SQL クエリ結果の個々の結果行を表すタプルに似たデータ構造です。行に "my_column" という名前の列が含まれている場合は、"my_column" を介して row の row.my_column フィールドにアクセスできます。数値インデックスを使用して、row[0] などのフィールドにアクセスすることもできます。列名が属性メソッド名として許可されていない場合 (たとえば、数字で始まる場合)、row["1_my_column"] としてフィールドにアクセスできます。

バージョン 1.0 以降

次のような Row メソッドがあります。

メソッド

メソッド	Description
`asDict`	フィールド名でインデックスが付けられた行の辞書表現を返します。重複するフィールド名がある場合は、重複するフィールドの 1 つ (ただし 1 つのみ) が辞書に返されます。どの重複フィールドが返されるかは定義されていません。

型の変換

次の表は、Apache Spark SQL データ型と、同等の Python データ型の対応付けを示しています。

Apache Spark SQL データ型	Python データ型
`array`	`numpy.ndarray`
`bigint`	`int`
`binary`	`bytearray`
`boolean`	`bool`
`date`	`datetime.date`
`decimal`	`decimal.Decimal`
`double`	`float`
`int`	`int`
`map`	`str`
`null`	`NoneType`
`smallint`	`int`
`string`	`str`
`struct`	`str`
`timestamp`	`datetime.datetime`
`tinyint`	`int`

テレメトリの収集

Databricks SQL Connector for Python は、Azure Databricks の信頼性の向上と問題のトラブルシューティングに役立つテレメトリデータを収集します。テレメトリは既定で有効になっており、次の操作データが収集されます。

ドライバーのバージョン、Python ランタイム、オペレーティングシステムなどのクライアント環境の詳細
ドライバー接続の構成 (個人を特定できる情報を除く)
操作の待機時間の測定
インライン JSON や Apache Arrow などの実行結果形式
クエリの実行、メタデータクエリ、ボリューム操作などの操作の種類
エラー分類データ
再試行回数

重要

Azure Databricks では、テレメトリを通じてクエリコンテンツ、クエリ結果、または個人を特定できる情報 (PII) は収集されません。

テレメトリ収集を無効にするには、接続の作成時に enable_telemetry パラメーターを 0 に設定します。

トラブルシューティング

`tokenAuthWrapperInvalidAccessToken: Invalid access token` メッセージ

問題: コードを実行すると、Error during request to server: tokenAuthWrapperInvalidAccessToken: Invalid access token のようなメッセージが表示されます。

考えられる原因: access_token に渡された値が、Azure 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 に渡される値が正しいことを確認して、もう一度試します。

サーバーホスト名の検索について詳しくは、「Azure Databricks コンピューティングリソースの接続の詳細を取得する」をご覧ください。

`IpAclError` メッセージ

問題: コードを実行すると、Azure Databricks ノートブックでコネクタを使用しようとしたときに Error during request to server: IpAclValidation メッセージが表示されます。

考えられる原因: Azure Databricks ワークスペースで IP 許可リストを有効にしている可能性があります。 IP 許可リストがあると、既定では Spark クラスターからコントロールプレーンへの接続が許可されません。

推奨される修正: コンピューティングプレーンサブネットを IP 許可リストに追加するよう管理者に依頼します。

その他のリソース

詳細については次を参照してください:

GitHub の Databricks SQL Connector for Python リポジトリ
データ型
Python Web サイトの組み込み型 (bool、bytearray、float、int、str について)
Python Web サイトの datetime (datetime.date および datatime.datetime について)
Python Web サイトのdecimal (decimal.Decimal について)
Python Web サイトの組み込み定数 (NoneType について)

フィードバック

このページはお役に立ちましたか?

Last updated on 2025-10-31

次の方法で共有

Python 用 Databricks SQL コネクタ

要件

概要

汎用コンピューティング

SQL ウェアハウス

認証

Databricks 個人用アクセス トークン認証

OAuth マシン間 (M2M) 認証

Microsoft Entra ID トークン認証

OAuth ユーザー マシン間 (U2M) 認証

例

User-Agent の設定

クエリ データ

クエリ タグ

データの挿入

クエリ メタデータ

カーソルと接続を管理する

Unity Catalog ボリューム内のファイルを管理する

ログの構成

テスト

API リファレンス

モジュール

connect メソッド

Connection クラス

メソッド

Cursor クラス

属性

メソッド

Row クラス

メソッド

型の変換

テレメトリの収集

トラブルシューティング

tokenAuthWrapperInvalidAccessToken: Invalid access token メッセージ

gaierror(8, 'nodename nor servname provided, or not known') メッセージ

IpAclError メッセージ

その他のリソース

フィードバック

その他のリソース

Databricks 個人用アクセストークン認証

OAuth ユーザーマシン間 (U2M) 認証

クエリデータ

クエリタグ

クエリメタデータ

`Connection` クラス

`Cursor` クラス

`Row` クラス

`tokenAuthWrapperInvalidAccessToken: Invalid access token` メッセージ

`gaierror(8, 'nodename nor servname provided, or not known')` メッセージ

`IpAclError` メッセージ