適用於 Python 的 Databricks SQL 連接器

適用於 Python 的 Databricks SQL 連接器是一個 Python 程式庫，可讓您使用 Python 程式碼在 Azure Databricks 叢集和 Databricks SQL 倉儲上執行 SQL 命令。 適用於 Python 的 Databricks SQL 連接器比類似的 Python 程式庫 (例如 pyodbc) 更容易設定和使用。 此程式庫遵循 PEP 249 – Python 資料庫 API 規格 v2.0。

需求

• 執行 Python >=3.8 和 <=3.11 的開發機器。
• Databricks 建議您使用 Python 虛擬環境，例如 Python 隨附的 venv 所提供的環境。 虛擬環境可協助您確保同時使用正確的 Python 版本和適用於 Python 的 Databricks SQL 連接器。 設定和使用虛擬環境不屬於本文的討論範圍。 如需詳細資訊，請參閱建立虛擬環境。
• 現有的叢集或 SQL 倉儲。

開始使用

• 透過執行 pip install databricks-sql-connector 或 python -m pip install databricks-sql-connector，在您的開發機器上安裝適用於 Python 的 Databricks SQL 連接器程式庫。

• 針對想要使用的叢集或 SQL 倉儲，收集下列資訊：

  Cluster

  • 叢集的伺服器主機名稱。 您可以從叢集的 [進階選項 > JDBC/ODBC] 索引標籤中的 [伺服器主機名稱] 值取得此值。
  • 叢集的 HTTP 路徑。 您可以從叢集的 [進階選項 > JDBC/ODBC] 索引標籤中的 [HTTP 路徑] 值取得此值。

  SQL 倉儲

  • SQL 倉儲的伺服器主機名稱。 您可以從 SQL 倉儲的 [連線詳細資料] 索引標籤中的 [伺服器主機名稱] 值取得此值。
  • SQL 倉儲的 HTTP 路徑。 您可以從 SQL 倉儲的 [連線詳細資料] 索引標籤中的 [HTTP 路徑] 值取得此值。

驗證

適用於 Python 的 Databricks SQL 連接器支援下列 Azure Databricks 驗證類型：

• Databricks 個人存取權杖驗證
• Microsoft Entra ID 權杖驗證
• OAuth 機器對機器 (M2M) 驗證
• OAuth 使用者對機器 (U2M) 驗證

適用於 Python 的 Databricks SQL 連接器尚不支援下列 Azure Databricks 驗證類型：

• Azure 受控識別驗證
• MS Entra 服務主體驗證
• Azure CLI 驗證

Databricks 個人存取權杖驗證

若要搭配 Azure Databricks 個人存取令牌驗證使用適用於 Python 的 Databricks SQL 連接器，您必須先建立 Azure Databricks 個人存取令牌。 若要這樣做，請遵循適用於工作區使用者的 Azure Databricks 個人存取令牌中的步驟。

若要驗證適用於 Python 的 Databricks SQL 連接器，請使用下列程式碼片段。 此片段假設您已設定下列環境變數：

• 將 DATABRICKS_SERVER_HOSTNAME 設定為叢集或 SQL 倉儲的 [伺服器主機名稱] 值。
• DATABRICKS_HTTP_PATH，設定為叢集或 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) 驗證

適用於 Python 的 Databricks SQL 連接器 2.7.0 版和更新版本支援 OAuth 機器對機器 (M2M) 驗證。 您也必須安裝適用於 Python 的 Databricks SDK 0.18.0 或更新版本 (例如透過執行 pip install databricks-sdk 或 python -m pip install databricks-sdk)。

若要搭配 OAuth M2M 驗證使用適用於 Python 的 Databricks SQL 連接器，您必須執行下列動作：

1. 在 Azure Databricks 工作區中建立 Azure Databricks 服務主體，並為該服務主體建立 OAuth 祕密。

   若要建立服務主體及其 OAuth 祕密，請參閱使用 OAuth (OAuth M2M) 透過服務主體對 Azure Databricks 的存取進行驗證。 記下服務主體的 [UUID] 或 [應用程式 ID] 值，以及服務主體 OAuth 祕密的 [祕密]值。

2. 為服務主體提供叢集或倉儲的存取權。

   若要授與服務主體對叢集或倉儲的存取權，請參閱計算權限或管理 SQL 倉儲。

若要驗證適用於 Python 的 Databricks SQL 連接器，請使用下列程式碼片段。 此片段假設您已設定下列環境變數：

• 將 DATABRICKS_SERVER_HOSTNAME 設定為叢集或 SQL 倉儲的 [伺服器主機名稱] 值。
• DATABRICKS_HTTP_PATH，設定為叢集或 SQL 倉儲的 HTTP 路徑 值。
• DATABRICKS_CLIENT_ID，設定為服務主體的 [UUID] 或 [應用程式 ID] 值。
• DATABRICKS_CLIENT_SECRET，設定為 Azure Databricks 服務主體的 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 權杖驗證

若要使用適用於 Python 的 Databricks SQL 連接器搭配 Microsoft Entra ID 權杖驗證，您必須提供適用於 Python 的 Databricks SQL 連接器和 Microsoft Entra ID 權杖。 若要建立 Microsoft Entra ID 存取權杖，請按照下列步驟操作：

• 針對 Azure Databricks 使用者，您可使用 Azure CLI。 請參閱使用 Azure CLI 取得使用者的 Microsoft Entra ID 權杖。
• 如需 Microsoft Entra ID 服務主體，請參閱使用 Azure CLI 取得Microsoft Entra ID 存取權杖。 若要建立 Microsoft Entra ID 受控服務主體，請參閱管理服務主體。

Microsoft Entra ID 權杖的預設存留期約為 1 小時。 若要建立新的 Microsoft Entra ID 權杖，請重複此程序。

若要驗證適用於 Python 的 Databricks SQL 連接器，請使用下列程式碼片段。 此片段假設您已設定下列環境變數：

• 將 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) 驗證

適用於 Python 的 Databricks SQL 連接器 2.7.0 版和更新版本支援 OAuth 使用者對機器 (M2M) 驗證。 您也必須安裝適用於 Python 的 Databricks SDK 0.19.0 或更新版本 (例如透過執行 pip install databricks-sdk 或 python -m pip install databricks-sdk)。

若要使用 OAuth U2M 驗證來驗證適用於 Python 的 Databricks SQL 連接器，請使用下列程式碼片段。 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:
# ...

範例

下列程式碼範例示範如何使用適用於 Python 的 Databricks SQL 連接器來查詢和插入資料、查詢中繼資料、管理資料指標和連線，以及設定記錄。

這些程式碼範例會從下列環境變數擷取其 server_hostname、http_path 和 access_token 連線變數值：

• DATABRICKS_SERVER_HOSTNAME，代表來自需求的 [伺服器主機名稱] 值。
• DATABRICKS_HTTP_PATH，代表來自需求的 [HTTP 路徑] 值。
• DATABRICKS_TOKEN，代表來自需求的存取權杖。

您可使用其他方法來擷取這些連線變數值。 使用環境變數只是眾多方法之一。

• 查詢資料
• 插入資料
• 查詢中繼資料
• 管理資料指標和連線
• 管理 Unity 目錄磁碟區中的檔案
• 設定記錄

查詢資料

下列程式碼範例示範如何呼叫適用於 Python 的 Databricks SQL 連接器，在叢集或 SQL 倉儲上執行基本 SQL 命令。 此命令會從 samples 目錄的 nyctaxi 結構描述中的 trips 資料表傳回前兩個資料列。

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

管理資料指標和連線

最佳做法是關閉不再使用的任何連線和資料指標。 這會釋放 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 目錄磁碟區中的檔案

Databricks SQL 連接器可讓您將本機檔案寫入 Unity 目錄磁碟區、從磁碟區下載檔案，以及從磁碟區刪除檔案，如下列範例所示：

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 Connector 會使用 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()

測試

若要測試程式碼，請使用 Python 測試架構，例如 pytest。 若要在不呼叫 Azure Databricks REST API 端點或變更 Azure Databricks 帳戶或工作區的狀態的情況下，在模擬條件下測試程式碼，您可以使用 Python 模擬程式庫，例如 unittest.mock。

例如，假設下列名為 helpers.py 的檔案，其中包含一個 get_connection_personal_access_token 函式，該函式使用 Azure Databricks 個人存取權杖傳回連線到 Azure Databricks 工作區，以及一個 select_nyctaxi_trips 函式，該函式使用連線從 samples 目錄 nyctaxi 結構描述中的 trips 資料表取得指定資料列數目：

# 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

並假設下列名為 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 參考

• 套件
• 模組
• 類別
• Connection 類別
• Cursor 類別
• Row 類別
• 型別轉換

套件

databricks-sql-connector

使用方式：pip install databricks-sql-connector

另請參閱 Python 套件索引 (PyPI) 中的 databricks-sql-connector。

模組

databricks.sql

使用方式：from databricks import sql

類別

選取的類別包括下列項目：

Connection 類別

若要建立 Connection 物件，使用下列參數呼叫 databricks.sql.connect 方法：

選取的 Connection 方法包括如下項目：

Cursor 類別

若要建立 Cursor 物件，請呼叫 Connection 類別的 cursor 方法。

選取的 Cursor 屬性包括下列項目：

選取的 Cursor 方法包括如下項目：

Row 類別

資料列類別是類似元組的資料結構，代表個別結果資料列。 如果資料列包含名稱為 "my_column" 的資料行，可以透過 row.my_column 存取 row 的 "my_column" 欄位。 您也可以使用數值索引來存取欄位，例如 row[0]。 如果不允許資料列名稱作為屬性方法名稱 (例如，以數位作開頭)，則您可以將欄位存取為 row["1_my_column"]。

自 1.0 版起

選取的 Row 方法包括：

| asDict

傳回以欄位名稱編製索引之資料列的字典表示。 如果有重複的欄位名稱，則會在字典中傳回其中一個重複的欄位 (但只有一個)。 未定義傳回重複欄位。

無參數。

傳回欄位的 dict。 |

類型轉換

下列資料表會將 Apache Spark SQL 資料類型對應至其 Python 資料類型對等項目。

疑難排解

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 上的適用於 Python 的 Databricks SQL 連接器存放庫
• 資料類型
• Python 網站上的內建類型 (適用於 bool、、bytearray、float、int和 str)
• Python 網站上的日期時間 (適用於 datetime.date 和 datatime.datetime)
• Python 網站上的 decimal (對於 decimal.Decimal)
• Python 網站上的內建常數 (對於 NoneType)