メインコンテンツまでスキップ

Python 用 Databricks SQL コネクタ

Databricks SQLConnector forPython は、Python PythonSQLコードを使用してDatabricks 汎用コンピュートおよびDatabricks SQL ウェアハウスで コマンドを実行できる ライブラリです。Databricks SQL Connector for Python は、 pyodbc などの同様の Python ライブラリよりもセットアップと使用が簡単です。このライブラリは、 PEP 249 – Python Database API Specification v2.0 に準拠しています。

important

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

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

必要条件​

  • Python 3.8 以上を実行する開発マシン。
  • Databricks では、Python に含まれている venv によって提供される環境など、Python 仮想環境を使用することをお勧めします。 仮想環境は、正しいバージョンの Python と Databricks SQL Connector for Python を一緒に使用していることを確認するのに役立ちます。 仮想環境の設定と使用については、この記事の範囲外です。 詳細については、「 仮想環境の作成」を参照してください。
  • 既存の 汎用コンピュート または SQLウェアハウス。

はじめに​

  1. Databricks SQL Connector for Python をインストールします。PyArrow は、Python 用の Databricks SQL コネクタのオプションの依存関係であり、コネクタのバージョン 4.0.0 以降では デフォルト によってインストールされません。PyArrow がインストールされていない場合、CloudFetch やその他の Apache Arrow 機能などの機能は利用できず、大量のデータのパフォーマンスに影響を与える可能性があります。

    • リーンコネクタを取り付けるには、以下を使用します。

      pip install databricks-sql-connector

    • PyArrow を含む完全なコネクタをインストールするには、次を使用します。

      pip install databricks-sql-connector[pyarrow]

  2. 使用する汎用コンピュートまたは SQLウェアハウスについて、次の情報を収集します。

  • 汎用コンピュートのサーバーホスト名。 これは、汎用コンピュートの 「Advanced OptionsJDBC ODBC > / 」タブの 「Server Hostname 」の値から取得できます。
  • 万能コンピュートのHTTPパス。 これは、汎用コンピュートの 「Advanced OptionsJDBC ODBC > / 」タブの 「HTTP Path 」の値から取得できます。
注記

The SQL connector does not support connecting to jobs compute.

認証​

Python 用 Databricks SQL コネクタでは、次の Databricks 認証の種類がサポートされています。

Python 用の Databricks SQL コネクタは、次の Databricks 認証の種類をまだサポートしていません。

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

Databricks個人アクセス VPN 認証でDatabricks SQL Connector for Pythonを使用するには、まずDatabricks個人アクセス トークンを作成する必要があります。 これを行うには、 「 ワークスペース ユーザー向けの個人アクセス墨を作成する 」のステップに従います。

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

  • DATABRICKS_SERVER_HOSTNAME汎用コンピュートまたはSQL ウェアハウスの サーバーホスト名 の値に設定します。
  • DATABRICKS_HTTP_PATHで、汎用コンピュートまたはSQL ウェアハウスの HTTP パス 値に設定します。
  • DATABRICKS_TOKENで、Databricks 個人用アクセス トークンに設定されます。

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

Python
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を実行します)。

OAuth M2M 認証で Databricks SQL Connector for Python を使用するには、次の操作を行う必要があります。

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

    サービスプリンシパルとそのOAuthシークレットを作成するには、「サービスプリンシパルのアクセスをOAuthでDatabricksに許可する」を参照してください。サービスプリンシパルの UUID または アプリケーション ID の値と、サービスプリンシパルのOAuthシークレットの Secret 値をメモします。

  2. そのサービスプリンシパルに、汎用コンピュートまたはウェアハウスへのアクセス権を付与します。

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

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

  • DATABRICKS_SERVER_HOSTNAME 汎用コンピュートまたはSQL ウェアハウスの サーバーホスト名 の値に設定します。
  • DATABRICKS_HTTP_PATHで、汎用コンピュートまたはSQL ウェアハウスの HTTP パス 値に設定します。
  • DATABRICKS_CLIENT_IDで、サービスプリンシパルの UUID または アプリケーション ID の値に設定します。
  • DATABRICKS_CLIENT_SECRETで、サービスプリンシパルの OAuth シークレットの Secret 値に設定します。

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

Python
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 バージョン 3.0.3 以上 は、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 認証では、リアルタイムの人間によるサインインと同意を使用して、ターゲットの Databricks ユーザー アカウントを認証します。 このスニペットは、次の環境変数を設定していることを前提としています。

  • DATABRICKS_SERVER_HOSTNAME を汎用コンピュートまたはウェアハウスの Server Hostname 値SQL設定します。
  • DATABRICKS_HTTP_PATH を HTTP Path value に設定して、汎用コンピュートまたはウェアハウスSQL。

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

Python
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 Catalog 内のファイルの管理、ログの構成を行う方法を示しています。

注記

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

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

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

User-Agent の設定​

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

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

Python
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テーブルに表示されます。

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

データの挿入​

次の例は、少量のデータ (数千行) を挿入する方法を示しています。

Python
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 コマンドを実行する必要があります。

クエリのメタデータ​

メタデータを取得するための専用のメソッドがあります。 次の例では、サンプルテーブルの列に関するメタデータを取得します。

Python
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を明示的に呼び出すことができます。

Python
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 ボリュームへのローカル ファイルの書き込み、ボリュームからのファイルのダウンロード、ボリュームからのファイルの削除を行うことができます。

Python
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 の 標準ログ モジュールが使用されます。次の例では、ログレベルを設定し、デバッグログを生成します。

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 という名前の次のファイルがあるとします。

Python
# 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

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

Python
# 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 つが期待値と一致するかどうかがチェックされます。

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

接続方式​

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

パラメーター

Type

説明

server_hostname

str

必須。サーバーホスト名は、万能コンピュートまたは SQLウェアハウス、たとえば 1234567890123456.7.gcp.databricks.com。

サーバーのホスト名を取得するには、「 はじめに」の手順を参照してください。

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

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]]]

随意。クライアントが行うすべての RPC 要求で HTTP ヘッダーに設定する追加 (キーと値のペア。 通常の使用法では、追加の HTTP ヘッダーは設定されません。デフォルトは Noneです。

catalog

str

随意。接続に使用する初期カタログ。デフォルトから None に設定します(この場合、 デフォルトカタログ、通常は hive_metastore が使用されます)。

schema

str

随意。接続に使用する初期スキーマ。デフォルトを None に設定します(その場合は、 デフォルトスキーマ default が使用されます)。

バージョン 2.0 から

use_cloud_fetch

bool

随意。データのチャンクをダウンロードするために、フェッチ要求をクラウドオブジェクトストアに直接送信するかどうか。デフォルトは Trueです。False に設定して、フェッチ要求を Databricks に直接送信します。

use_cloud_fetch が True に設定されているのにネットワークアクセスがブロックされている場合、フェッチリクエストは失敗します。

バージョン 2.8 から

user_agent_entry

str

随意。使用状況の追跡のために HTTP 要求ヘッダーに含める User-Agent エントリ。デフォルトは PyDatabricksSqlConnectorです。

Connection クラス​

コンピュートや SQLウェアハウスとのつながりを表します。

メソッド​

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

手法

説明

close

データベースへの接続を閉じ、サーバー上のすべての関連リソースを解放します。 この接続に対して追加の呼び出しを行うと、 Error.

パラメーターはありません。

戻り値はありません。

cursor

データベース内のレコードのトラバースを可能にする新しい Cursor オブジェクト を返します。

パラメーターはありません。

Cursorクラス​

データ・レコードをトラバースするためのメカニズムを表します。

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

属性​

選択した Cursor 属性には次のものが含まれます。

属性

説明

arraysize

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

description

tuple 個のオブジェクトの Python listが含まれています。これらの tuple オブジェクトにはそれぞれ 7 つの値が含まれ、各 tuple オブジェクトの最初の 2 つのアイテムには、次のように 1 つの結果列を記述する情報が含まれています。

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

メソッド​

選択された Cursor 方法は次のとおりです。

手法

説明

cancel

カーソルが開始したデータベースクエリまたはコマンドの実行を中断します。サーバー上の関連リソースを解放するには、cancel メソッドを呼び出した後に close メソッドを呼び出します。

パラメーターはありません。

戻り値はありません。

close

カーソルを閉じ、サーバー上の関連リソースを解放します。 既に閉じているカーソルを閉じると、エラーがスローされる可能性があります。

パラメーターはありません。

戻り値はありません。

execute

データベース・クエリまたはコマンドを準備して実行します。

パラメーター:

  • operation：必須。準備して実行するクエリまたはコマンド。種類： str

parameters パラメーターを使用しない例:

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 of 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

クエリのすべての (または残りのすべての) 行を取得します。

パラメーターはありません。

クエリのすべての (または残りのすべての) 行を、Rowオブジェクトの Python listとして返します。

前回の execute メソッドの呼び出しでデータが返されなかった場合、または execute 呼び出しがまだ行われていない場合は、Error を投げます。

fetchmany

クエリの次の行を取得します。

パラメーター:

  • size：随意。取得する次の行の数。指定しない場合は、 arraysize 属性の値が使用されます。タイプ: int。

例： cursor.fetchmany(10)

クエリの次の行の最大 size (または、sizeが指定されていない場合は arraysize 属性) を、Row オブジェクトの Python listとして返します。

フェッチする行が size 行未満の場合、残りのすべての行が返されます。

前回の execute メソッドの呼び出しでデータが返されなかった場合、または execute 呼び出しがまだ行われていない場合は、Error を投げます。

fetchone

データセットの次の行を取得します。

パラメーターはありません。

データセットの次の行を Python tuple オブジェクトとして 1 つのシーケンスとして返すか、使用可能なデータがなくなった場合は None を返します。

前回の execute メソッドの呼び出しでデータが返されなかった場合、または execute 呼び出しがまだ行われていない場合は、Error を投げます。

fetchall_arrow

クエリのすべての (または残りのすべての) 行を PyArrow Table オブジェクトとして取得します。 非常に大量のデータを返すクエリでは、メモリ消費量を減らすために、代わりに fetchmany_arrow を使用する必要があります。

パラメーターはありません。

クエリの全行（または残りの全行）をPyArrowテーブルとして返します。

前回の execute メソッドの呼び出しでデータが返されなかった場合、または execute 呼び出しがまだ行われていない場合は、Error を投げます。

バージョン 2.0 から

fetchmany_arrow

クエリの次の行を PyArrow Table オブジェクトとして取得します。

パラメーター:

  • size：随意。取得する次の行の数。指定しない場合は、 arraysize 属性の値が使用されます。タイプ: int。

例： cursor.fetchmany_arrow(10)

クエリの次の行の size 引数 (指定されていない場合は arraysize 属性) まで size Python PyArrow Table オブジェクトとして返します。

前回の execute メソッドの呼び出しでデータが返されなかった場合、または execute 呼び出しがまだ行われていない場合は、Error を投げます。

バージョン 2.0 から

Row クラス​

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

バージョン1.0以降

選択された Row 方法は次のとおりです。

メソッド​

手法

説明

asDict

フィールド名で索引付けされた行の辞書表現を返します。 フィールド名が重複している場合、重複したフィールドの1つ（ただし、1つのみ）が辞書に返されます。 どの重複フィールドが返されるかは定義されていません。

型変換​

次の表は、Apache Spark SQL データ型を Python データ型と同等の 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

トラブルシューティング​

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 コンピュート リソースの接続の詳細を取得する」を参照してください。

IpAclError メッセージ​

問題 : コードを実行すると Error during request to server: IpAclValidation 、 Databricks ノートブック上のコネクタ。

考えられる原因 : Databricks ワークスペースで IP 許可リストが有効になっている可能性があります。 IP 許可リスト、接続 Spark クラスターからコントロールプレーンに戻ることは、デフォルトでは許可されていません。

推奨される修正: 管理者に依頼して、コンピュート プレーン サブネットを IP 許可リストに追加してください。

追加のリソース​

詳細については、以下を参照してください。

に最終更新