Utilizzo avanzato di Databricks Connect per Python

Questo articolo descrive gli argomenti che vanno oltre la configurazione di base di Databricks Connect.

Configurare l'stringa di connessione Spark Connect

Oltre a connettersi al cluster usando le opzioni descritte in Configurare una connessione a un cluster, un'opzione più avanzata consiste nel connettersi usando l'stringa di connessione Spark Connect. È possibile passare la stringa nella remote funzione o impostare la SPARK_REMOTE variabile di ambiente.

Per impostare il stringa di connessione usando la remote funzione :

# Set the Spark Connect connection string in DatabricksSession.builder.remote.
from databricks.connect import DatabricksSession

workspace_instance_name = retrieve_workspace_instance_name()
token                   = retrieve_token()
cluster_id              = retrieve_cluster_id()

spark = DatabricksSession.builder.remote(
   f"sc://{workspace_instance_name}:443/;token={token};x-databricks-cluster-id={cluster_id}"
).getOrCreate()

In alternativa, impostare la SPARK_REMOTE variabile di ambiente:

sc://<workspace-instance-name>:443/;token=<access-token-value>;x-databricks-cluster-id=<cluster-id>

Inizializzare quindi la DatabricksSession classe come segue:

from databricks.connect import DatabricksSession

spark = DatabricksSession.builder.getOrCreate()

Shell Pyspark

Databricks Connect per Python viene fornito con un pyspark file binario che è un PYSpark REPL (una shell Spark) configurato per l'uso di Databricks Connect.

Quando si inizia senza parametri aggiuntivi, la shell preleva le credenziali predefinite dall'ambiente (ad esempio, le DATABRICKS_ variabili di ambiente o il DEFAULT profilo di configurazione) per connettersi al cluster Azure Databricks. Per informazioni sulla configurazione di una connessione, vedere Configurazione di calcolo per Databricks Connect.

1. Per avviare la shell Spark e connetterla al cluster in esecuzione, eseguire uno dei comandi seguenti dall'ambiente virtuale Python attivato:

   pyspark
   

   Viene visualizzata la shell Spark, ad esempio:

   Python 3.10 ...
   [Clang ...] on darwin
   Type "help", "copyright", "credits" or "license" for more information.
   Welcome to
        ____              __
       / __/__  ___ _____/ /__
      _\ \/ _ \/ _ `/ __/  '_/
     /__ / .__/\_,_/_/ /_/\_\   version 13.x.dev0
        /_/
   
   Using Python version 3.10 ...
   Client connected to the Spark Connect server at sc://...:.../;token=...;x-databricks-cluster-id=...
   SparkSession available as 'spark'.
   >>>
   

   Dopo l'avvio della shell, l'oggetto spark è disponibile per eseguire comandi Apache Spark nel cluster Databricks. Eseguire un semplice comando PySpark, ad esempio spark.range(1,10).show(). Se non sono presenti errori, la connessione è stata completata correttamente.

2. Per informazioni su come usare la shell Spark con Python per eseguire comandi nel calcolo, vedere Interactive Analysis with the Spark Shell with Python (Analisi interattiva con Spark Shell con Spark Shell).

   Usare la variabile predefinita spark per rappresentare nel SparkSession cluster in esecuzione, ad esempio:

   >>> df = spark.read.table("samples.nyctaxi.trips")
   >>> df.show(5)
   +--------------------+---------------------+-------------+-----------+----------+-----------+
   |tpep_pickup_datetime|tpep_dropoff_datetime|trip_distance|fare_amount|pickup_zip|dropoff_zip|
   +--------------------+---------------------+-------------+-----------+----------+-----------+
   | 2016-02-14 16:52:13|  2016-02-14 17:16:04|         4.94|       19.0|     10282|      10171|
   | 2016-02-04 18:44:19|  2016-02-04 18:46:00|         0.28|        3.5|     10110|      10110|
   | 2016-02-17 17:13:57|  2016-02-17 17:17:55|          0.7|        5.0|     10103|      10023|
   | 2016-02-18 10:36:07|  2016-02-18 10:41:45|          0.8|        6.0|     10022|      10017|
   | 2016-02-22 14:14:41|  2016-02-22 14:31:52|         4.51|       17.0|     10110|      10282|
   +--------------------+---------------------+-------------+-----------+----------+-----------+
   only showing top 5 rows
   

   Tutto il codice Python viene eseguito localmente, mentre tutto il codice PySpark che coinvolge le operazioni del dataframe viene eseguito nel cluster nell'area di lavoro remota di Azure Databricks e le risposte di esecuzione vengono inviate al chiamante locale.

3. Per arrestare la shell spark, premere Ctrl + d o Ctrl + zoppure eseguire il comando quit() o exit().

Intestazioni HTTP aggiuntive

Databricks Connect comunica con i cluster Databricks tramite gRPC su HTTP/2.

Alcuni utenti avanzati possono scegliere di installare un servizio proxy tra il client e il cluster Azure Databricks, per avere un migliore controllo sulle richieste provenienti dai client.

I proxy, in alcuni casi, possono richiedere intestazioni personalizzate nelle richieste HTTP.

Il headers() metodo può essere usato per aggiungere intestazioni personalizzate alle richieste HTTP.

spark = DatabricksSession.builder.header('x-custom-header', 'value').getOrCreate()

Certificati

Se il cluster si basa su un certificato SSL/TLS personalizzato per risolvere un nome di dominio completo (FQDN) di un'area di lavoro di Azure Databricks, è necessario impostare la variabile GRPC_DEFAULT_SSL_ROOTS_FILE_PATH di ambiente nel computer di sviluppo locale. Questa variabile di ambiente deve essere impostata sul percorso completo del certificato installato nel cluster.

Ad esempio, si imposta questa variabile di ambiente nel codice Python come indicato di seguito:

import os

os.environ["GRPC_DEFAULT_SSL_ROOTS_FILE_PATH"] = "/etc/ssl/certs/ca-bundle.crt"

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

Registrazione e debug dei log

Databricks Connect per Python produce log usando la registrazione Python standard.

I log vengono generati nel flusso di errore standard (stderr) e per impostazione predefinita vengono generati solo i log a livello DI AVVISO e versioni successive.

L'impostazione di una variabile SPARK_CONNECT_LOG_LEVEL=debug di ambiente modifica questa impostazione predefinita e stampa tutti i messaggi di log al DEBUG livello e superiore.