Connettore SQL di Databricks per Python

Il connettore SQL di Databricks per Python è una libreria Python che consente di usare il codice Python per eseguire comandi SQL nei cluster di Azure Databricks e nei databricks SQL Warehouse. Il connettore SQL di Databricks per Python è più semplice da configurare e usare rispetto a librerie Python simili, ad esempio pyodbc. Questa libreria segue la Specifica dell'API di Database di Python PEP 249 v2.0.

Requisiti

• Un computer di sviluppo che esegue Python >=3.8 e <=3.11.
• Databricks consiglia di usare ambienti virtuali Python, ad esempio quelli forniti da venv inclusi in Python. Gli ambienti virtuali consentono di assicurarsi di usare insieme le versioni corrette di Python e il connettore SQL di Databricks per Python. La configurazione e l'uso di ambienti virtuali non rientra nell'ambito di questo articolo. Per altre informazioni, consultare Creare ambienti virtuali.
• Un cluster esistente o SQL warehouse.

Inizia

• Installare il connettore SQL di Databricks per Python. pyArrow è una dipendenza facoltativa del connettore SQL di Databricks per Python e non è installata per impostazione predefinita. Se non si installa PyArrow, le funzionalità come CloudFetch e altre funzionalità di Apache Arrow non sono disponibili, che potrebbero influire sulle prestazioni per grandi volumi di dati.

  • Per installare il connettore snello, usare pip install databricks-sql-connector.
  • Per installare il connettore completo, incluso PyArrow, usare pip install databricks-sql-connector[pyarrow].

• Raccogliere le informazioni seguenti per il cluster o SQL Warehouse da usare:

  Gruppo

  • Nome host del server del cluster. È possibile ottenerlo dal valore Hostname del server nella scheda Opzioni avanzate > JDBC/ODBC per il cluster.
  • Percorso HTTP del cluster. Puoi ottenere questo dal valore Percorso HTTP nella scheda Opzioni Avanzate > JDBC/ODBC per il tuo cluster.

  Magazzino SQL

  • Nome host del server di SQL Warehouse. È possibile ottenere questo dal valore nome host del server nella scheda Dettagli della connessione per il magazzino SQL.
  • Percorso HTTP del warehouse SQL. È possibile ottenerlo dal valore percorso HTTP nella scheda Dettagli connessione per il SQL warehouse.

Autenticazione

Il connettore del database SQL per Python supporta i seguenti tipi di autenticazione di Azure Databricks:

• Autenticazione con token di accesso personale di Databricks
• Autenticazione con token di Microsoft Entra ID
• Autenticazione OAuth da computer a computer (M2M)
• Autenticazione da utente a computer (U2M) OAuth

Il connettore SQL di Databricks per Python non supporta ancora i tipi di autenticazione di Azure Databricks seguenti:

• Autenticazione delle identità gestite di Azure
• Autenticazione principale di servizio di MS Entra
• Autenticazione con interfaccia della riga di comando di Azure

Autenticazione con token di accesso personale di Databricks

Per usare il connettore SQL di Databricks per Python con l'autenticazione del token di accesso personale di Azure Databricks, è prima necessario creare un token di accesso personale di Azure Databricks. A tale scopo, seguire la procedura descritta in Token di accesso personale di Azure Databricks per gli utenti dell'area di lavoro.

Per autenticare il connettore SQL di Databricks per Python, usare il frammento di codice seguente. Questo frammento presuppone che siano state impostate le variabili di ambiente seguenti:

• impostato sul valore Server Hostname per il cluster o SQL Warehouse.
• , impostare su valore del percorso HTTP per il cluster o il magazzino SQL.
• DATABRICKS_TOKEN, impostato sul token di accesso personale di Azure Databricks.

Per impostare le variabili di ambiente, vedere la documentazione del sistema operativo.

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

Autenticazione OAuth da computer a computer (M2M)

Databricks SQL Connector for Python versioni 2.7.0 e successive supporta l'autenticazione da computer a computer (M2M) OAuth. È anche necessario installare Databricks SDK per Python 0.18.0 o versione successiva, ad esempio eseguendo pip install databricks-sdk o python -m pip install databricks-sdk.

Per usare il connettore SQL di Databricks per Python con l'autenticazione OAuth M2M, è necessario eseguire le operazioni seguenti:

1. Creare un primario di servizio di Azure Databricks nell'area di lavoro di Azure Databricks e creare un segreto OAuth per tale primario di servizio.

   Per creare l'entità servizio e il relativo segreto OAuth, vedere Autorizzare l'accesso non presidiato alle risorse di Azure Databricks con un'entità servizio usando OAuth. Prendere nota del valore UUID o ID applicazione dell'entità servizio e del valore Secret per il segreto OAuth dell'entità servizio.

2. Concedere a quel principale del servizio l'accesso al cluster o al data warehouse.

   Per concedere al principal del servizio l'accesso al cluster o al SQL warehouse, vedere Autorizzazioni di calcolo o Gestire un data warehouse SQL.

Per autenticare il connettore SQL di Databricks per Python, usare il frammento di codice seguente. Questo frammento presuppone che siano state impostate le variabili di ambiente seguenti:

• DATABRICKS_SERVER_HOSTNAME impostato al valore Nome Host del Server per il tuo cluster o SQL Warehouse.
• DATABRICKS_HTTP_PATH, impostare il valore su Percorso HTTP per il cluster o il magazzino SQL.
• DATABRICKS_CLIENT_ID, impostato sull'UUID dell'entità principale del servizio o sul valore dell'ID applicazione.
• DATABRICKS_CLIENT_SECRET, impostato sul valore segreto per il segreto OAuth del principale del servizio.

Per impostare le variabili di ambiente, vedere la documentazione del sistema operativo.

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

Autenticazione con token di Microsoft Entra ID

Per usare il connettore SQL di Databricks per Python con l'autenticazione del token ID Microsoft Entra, è necessario fornire il connettore SQL databricks per Python con il token ID Microsoft Entra. Per creare un token di accesso di Microsoft Entra ID, eseguire le operazioni seguenti:

• Per un utente di Azure Databricks, è possibile usare l'interfaccia della riga di comando di Azure. Consulta Ottenere i token ID di Microsoft Entra per gli utenti utilizzando l'Azure CLI.
• Per un'entità di servizio di Microsoft Entra ID, consulta Ottenere un token di accesso di Microsoft Entra ID con Azure CLI. Per creare un'entità servizio gestita di Microsoft Entra ID, consultare Gestire le entità servizio.

I token ID Microsoft Entra hanno una durata predefinita di circa 1 ora. Per creare un nuovo token ID Microsoft Entra, ripetere questo processo.

Per autenticare il connettore SQL di Databricks per Python, usare il frammento di codice seguente. Questo frammento presuppone che siano state impostate le variabili di ambiente seguenti:

• Impostare DATABRICKS_SERVER_HOSTNAME sul valore Nome host del server per il cluster o magazzino SQL.
• Imposta il valore del percorso HTTP per il tuo cluster o magazzino SQL su DATABRICKS_HTTP_PATH.
• Impostare DATABRICKS_TOKEN sul token ID di Microsoft Entra.

Per impostare le variabili di ambiente, vedere la documentazione del sistema operativo.

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

Autenticazione da utente a computer (U2M) OAuth

Databricks SQL Connector per Python versioni 2.7.0 e successive supportano l'autenticazione OAuth da utente a macchina (U2M). È anche necessario installare Databricks SDK per Python 0.19.0 o versione successiva, ad esempio eseguendo pip install databricks-sdk o python -m pip install databricks-sdk.

Per autenticare il connettore SQL di Databricks per Python con l’autenticazione OAuth U2M, usare il frammento di codice seguente. L'autenticazione U2M OAuth utilizza l'accesso umano in tempo reale e il consenso per autenticare l'account utente di Azure Databricks di destinazione. Questo frammento presuppone che siano state impostate le variabili di ambiente seguenti:

• Impostare DATABRICKS_SERVER_HOSTNAME sul valore Nome host del server per il cluster o deposito SQL.
• Imposta DATABRICKS_HTTP_PATH sul valore del percorso HTTP per il tuo cluster o magazzino SQL.

Per impostare le variabili di ambiente, vedere la documentazione del sistema operativo.

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

Esempi

Gli esempi di codice seguenti illustrano come usare databricks SQL Connector per Python per eseguire query e inserire dati, eseguire query sui metadati, gestire cursori e connessioni e configurare la registrazione.

Questo esempio di codice recupera i valori delle variabili di connessione server_hostname, http_pathe access_token da queste variabili di ambiente:

• DATABRICKS_SERVER_HOSTNAME: che rappresenta il valore Server Hostname in base ai requisiti.
• DATABRICKS_HTTP_PATH: che rappresenta il valore Percorso HTTP in base ai requisiti.
• DATABRICKS_TOKEN, che rappresenta il token di accesso secondo i requisiti.

È possibile usare altri approcci per recuperare questi valori delle variabili di connessione. L'uso delle variabili di ambiente è solo un approccio tra molti.

• Impostare agente utente
• Interrogare dati
• Inserisci dati
• Metadati delle query
• Gestire cursori e connessioni
• Gestire i file nei volumi di Unity Catalog
• Configurare la registrazione

Impostare User-Agent

Nell'esempio di codice seguente viene illustrato come impostare l'applicazione User-Agent per il monitoraggio dell'utilizzo 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)

Eseguire query sui dati

L'esempio di codice seguente illustra come chiamare il connettore SQL di Databricks per Python per eseguire un comando SQL di base in un cluster o in un SQL warehouse. Questo comando restituisce le prime due righe della tabella trips nello schema samples del catalogo nyctaxi.

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)

Inserire dati

Nell'esempio seguente viene illustrato come inserire piccole quantità di dati (migliaia di righe):

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)

Per grandi quantità di dati, è necessario prima caricare i dati nell'archiviazione cloud e quindi eseguire il comando COPY INTO.

Metadati delle query

Sono disponibili metodi dedicati per il recupero dei metadati. Nell'esempio seguente vengono recuperati i metadati relativi alle colonne in una tabella di esempio:

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

Gestire cursori e connessioni

È consigliabile chiudere tutte le connessioni e i cursori che non sono più in uso. In questo modo vengono liberate le risorse nei cluster di Azure Databricks e nei data warehouse SQL di Databricks.

È possibile usare un gestore di contesto (la sintassi with usata negli esempi precedenti) per gestire le risorse o chiamare close in modo esplicito:

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

Gestire i file nei volumi del catalogo di Unity

Il connettore SQL di Databricks consente di scrivere file locali in volumi di Unity Catalog , scaricare file dai volumi ed eliminare file dai volumi, come illustrato nell'esempio seguente:

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'"
    )

Configurare la registrazione

Il connettore SQL di Databricks usa il modulo di registrazione standard di Python. È possibile configurare il livello di registrazione simile al seguente:

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

Collaudo

Per testare il codice, usare framework di test Python come pytest. Per testare il codice in condizioni simulate senza chiamare gli endpoint dell'API REST di Azure Databricks o modificare lo stato degli account o delle aree di lavoro di Azure Databricks, è possibile usare librerie di simulazione di Python, ad esempio le unittest.mock.

Ad esempio, dato il file seguente denominato helpers.py contenente una funzione get_connection_personal_access_token che usa un token di accesso personale di Azure Databricks per restituire una connessione a un'area di lavoro di Azure Databricks e una funzione select_nyctaxi_trips che usa la connessione per ottenere il numero specificato di righe di dati dalla tabella trips nello schema samples del catalogo 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(f"SELECT * FROM samples.nyctaxi.trips LIMIT {num_rows}")
  result: List[Row] = cursor.fetchall()
  return result

E dato il file seguente denominato main.py che chiama la funzione get_connection_personal_access_token e 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)

Il file seguente denominato test_helpers.py verifica se la funzione select_nyctaxi_trips restituisce la risposta prevista. Anziché creare una connessione reale all'area di lavoro di destinazione, questo test simula un oggetto Connection. Il test simula anche alcuni dati conformi allo schema e ai valori presenti nei dati reali. Il test restituisce i dati fittizi tramite la connessione fittizia e quindi controlla se uno dei valori delle righe di dati fittizi corrisponde al valore previsto.

# 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

Poiché la funzione select_nyctaxi_trips contiene un'istruzione SELECT e pertanto non modifica lo stato della tabella trips, la simulazione non è assolutamente necessaria in questo esempio. Tuttavia, la simulazione consente di eseguire rapidamente i test senza attendere che venga stabilita una connessione effettiva con l'area di lavoro. Inoltre, la simulazione consente di eseguire test simulati più volte per le funzioni che potrebbero modificare lo stato di una tabella, ad esempio INSERT INTO, UPDATEe DELETE FROM.

Informazioni di riferimento sulle API

• Pacchetto
• Modulo
• Classi
• Connection classe
• Classe Cursor
• ClasseRow
• Conversioni di tipi

Pacchetto

databricks-sql-connector

Utilizzo: pip install databricks-sql-connector

Consultare anche databricks-sql-connector nell'indice dei pacchetti Python (PyPI).

Modulo

databricks.sql

Utilizzo: from databricks import sql

Classi

Le classi selezionate includono quanto segue:

Classe Connection

Per creare un oggetto Connection, chiamare il metodo databricks.sql.connect con i parametri seguenti:

I metodi Connection selezionati includono i seguenti:

Classe Cursor

Per creare un oggetto Cursor, chiamare il metodo Connection della classe cursor.

Gli attributi selezionati Cursor includono quanto segue:

I metodi Cursor selezionati includono i seguenti:

Classe Row

La classe row è una struttura di dati simile ad una tupla che rappresenta una singola riga di risultato. Se la riga contiene una colonna con il nome "my_column", è possibile accedere al campo "my_column" di row tramite row.my_column. È anche possibile usare indici numerici per accedere ai campi, ad esempio row[0]. Se il nome della colonna non è consentito come nome del metodo di attributo (ad esempio, inizia con una cifra), è possibile accedere al campo come row["1_my_column"].

A partire dalla versione 1.0

I metodi Row selezionati includono:

| asDict

Restituisce una rappresentazione del dizionario della riga, indicizzata in base ai nomi dei campi. Se sono presenti nomi di campo duplicati, uno dei campi duplicati (ma solo uno) verrà restituito nel dizionario. Il campo duplicato restituito non è definito.

Nessun parametro.

Restituisce un dict relativo ai campi. |

Conversioni di tipi

La tabella seguente esegue il mapping dei tipi di dati SQL di Apache Spark agli equivalenti del tipo di dati Python.

Risoluzione dei problemi

tokenAuthWrapperInvalidAccessToken: Invalid access token messaggio

Problema: quando si esegue il codice, viene visualizzato un messaggio simile a Error during request to server: tokenAuthWrapperInvalidAccessToken: Invalid access token.

Possibile causa: il valore passato a access_token non è un token di accesso personale di Azure Databricks valido.

Correzione consigliata: verificare che il valore passato a access_token sia corretto e riprovare.

gaierror(8, 'nodename nor servname provided, or not known') messaggio

Problema: quando si esegue il codice, viene visualizzato un messaggio simile a Error during request to server: gaierror(8, 'nodename nor servname provided, or not known').

Possibile causa: il valore passato a server_hostname non è il nome host corretto.

Correzione consigliata: verificare che il valore passato a server_hostname sia corretto e riprovare.

Per altre informazioni su come trovare il nome host del server, vedere Ottenere i dettagli della connessione per una risorsa di calcolo di Azure Databricks.

IpAclError messaggio

Problema: quando si esegue il codice, viene visualizzato il messaggio Error during request to server: IpAclValidation quando si tenta di usare il connettore in un notebook di Azure Databricks.

Possibile causa: è possibile che l'elenco indirizzi IP consentiti sia abilitato per l'area di lavoro di Azure Databricks. Con l'elenco degli indirizzi IP consentiti, le connessioni dai cluster Spark al piano di controllo non sono consentite per impostazione predefinita.

correzione consigliata: chiedere all'amministratore di aggiungere la subnet del piano di calcolo all'elenco indirizzi IP consentiti.

Risorse aggiuntive

Per altre informazioni, vedi:

• Il repository del connettore SQL di Databricks per Python in GitHub
• Tipo di dati
• Tipi predefiniti (per bool, bytearray, float, int e str) nel sito Web Python
• datetime (per datetime.date e datatime.datetime) nel sito Web Python
• decimal (per decimal.Decimal) nel sito Web Python
• Costanti predefinite (per NoneType) nel sito Web Python