Usare l'API Livy per inviare ed eseguire processi di sessione

Si applica a:✅ Fabric Data Engineering and Data Science

Informazioni su come inviare processi di sessione Spark usando l'API Livy per Fabric Data Engineering.

Prerequisiti

• Fabric Premium o Capacità triale con Lakehouse

• Un client remoto, ad esempio Visual Studio Code con Jupyter Notebook, PySpark e Libreria di Autenticazione Microsoft (MSAL) per Python

• O un token dell'app Microsoft Entra. Registrare un'applicazione con il Microsoft Identity Platform

• Oppure un token SPN (Service Principal Name) di Microsoft Entra. Aggiungi e gestisci le credenziali dell'applicazione in Microsoft Entra

• Alcuni dati nel lakehouse, questo esempio usa NYC Taxi & Limousine Commission green_tripdata_2022_08 un file parquet caricato nella lakehouse

L'API Livy definisce un endpoint unificato per le operazioni. Sostituire i segnaposto {Entra_TenantID}, {Entra_ClientID}, {Fabric_WorkspaceID}, {Fabric_LakehouseID} con i valori appropriati quando si seguono gli esempi in questo articolo.

Configurare Visual Studio Code per la sessione dell'API Livy

1. Selezionare Lakehouse Settings nel tuo Fabric Lakehouse.

   

2. Passare alla sezione Livy endpoint.

   

3. Copiare la stringa di connessione del processo di sessione (prima casella rossa nell'immagine) nel codice.

4. Vai a Interfaccia di amministrazione di Microsoft Entra e copia sia l'ID applicazione (client) che l'ID directory (tenant) nel tuo codice.

   

Autenticare una sessione Spark dell'API Livy usando un token utente Microsoft Entra o un token SPN Microsoft Entra

Autenticare una sessione spark dell'API Livy usando un token SPN Microsoft Entra

1. Creare un notebook .ipynb in Visual Studio Code e inserire il codice seguente.

   import sys
   from msal import ConfidentialClientApplication
   
   # Configuration - Replace with your actual values
   tenant_id = "Entra_TenantID"  # Microsoft Entra tenant ID
   client_id = "Entra_ClientID"  # Service Principal Application ID
   
   # Certificate paths - Update these paths to your certificate files
   certificate_path = "PATH_TO_YOUR_CERTIFICATE.pem"      # Public certificate file
   private_key_path = "PATH_TO_YOUR_PRIVATE_KEY.pem"      # Private key file
   certificate_thumbprint = "YOUR_CERTIFICATE_THUMBPRINT" # Certificate thumbprint
   
   # OAuth settings
   audience = "https://analysis.windows.net/powerbi/api/.default"
   authority = f"https://login.windows.net/{tenant_id}"
   
   def get_access_token(client_id, audience, authority, certificate_path, private_key_path, certificate_thumbprint=None):
       """
       Get an app-only access token for a Service Principal using OAuth 2.0 client credentials flow.
   
       This function uses certificate-based authentication which is more secure than client secrets.
   
       Args:
           client_id (str): The Service Principal's client ID  
           audience (str): The audience for the token (resource scope)
           authority (str): The OAuth authority URL
           certificate_path (str): Path to the certificate file (.pem format)
           private_key_path (str): Path to the private key file (.pem format)
           certificate_thumbprint (str): Certificate thumbprint (optional but recommended)
   
       Returns:
           str: The access token for API authentication
   
       Raises:
           Exception: If token acquisition fails
       """
       try:
           # Read the certificate from PEM file
           with open(certificate_path, "r", encoding="utf-8") as f:
               certificate_pem = f.read()
   
           # Read the private key from PEM file
           with open(private_key_path, "r", encoding="utf-8") as f:
               private_key_pem = f.read()
   
           # Create the confidential client application
           app = ConfidentialClientApplication(
               client_id=client_id,
               authority=authority,
               client_credential={
                   "private_key": private_key_pem,
                   "thumbprint": certificate_thumbprint,
                   "certificate": certificate_pem
               }
           )
   
           # Acquire token using client credentials flow
           token_response = app.acquire_token_for_client(scopes=[audience])
   
           if "access_token" in token_response:
               print("Successfully acquired access token")
               return token_response["access_token"]
           else:
               raise Exception(f"Failed to retrieve token: {token_response.get('error_description', 'Unknown error')}")
   
       except FileNotFoundError as e:
           print(f"Certificate file not found: {e}")
           sys.exit(1)
       except Exception as e:
           print(f"Error retrieving token: {e}", file=sys.stderr)
           sys.exit(1)
   
   # Get the access token
   token = get_access_token(client_id, audience, authority, certificate_path, private_key_path, certificate_thumbprint)
   

2. Esegui la cella del notebook. Verrà visualizzato il token Microsoft Entra restituito.

   

Autenticare una sessione spark dell'API Livy usando un token utente Microsoft Entra

1. Creare un notebook .ipynb in Visual Studio Code e inserire il codice seguente.

   from msal import PublicClientApplication
   import requests
   import time
   
   # Configuration - Replace with your actual values
   tenant_id = "Entra_TenantID"  # Microsoft Entra tenant ID
   client_id = "Entra_ClientID"  # Application ID (can be the same as above or different)
   
   # Required scopes for Livy API access
   scopes = [
       "https://api.fabric.microsoft.com/Lakehouse.Execute.All",      # Required — execute operations in lakehouses
       "https://api.fabric.microsoft.com/Lakehouse.Read.All",         # Required — read lakehouse metadata
       "https://api.fabric.microsoft.com/Code.AccessFabric.All",      # Required — general Fabric API access from Spark Runtime
       "https://api.fabric.microsoft.com/Code.AccessStorage.All",     # Required — access OneLake and Azure storage from Spark Runtime
   ]
   
   # Optional scopes — add these only if your Spark jobs need access to the corresponding services:
   #    "https://api.fabric.microsoft.com/Code.AccessAzureKeyvault.All"     # Optional — access Azure Key Vault from Spark Runtime
   #    "https://api.fabric.microsoft.com/Code.AccessAzureDataLake.All"     # Optional — access Azure Data Lake Storage Gen1 from Spark Runtime
   #    "https://api.fabric.microsoft.com/Code.AccessAzureDataExplorer.All" # Optional — access Azure Data Explorer from Spark Runtime
   #    "https://api.fabric.microsoft.com/Code.AccessSQL.All"               # Optional — access Azure SQL audience tokens from Spark Runtime
   
   def get_access_token(tenant_id, client_id, scopes):
       """
       Get an access token using interactive authentication.
   
       This method will open a browser window for user authentication.
   
       Args:
           tenant_id (str): The Microsoft Entra tenant ID
           client_id (str): The application client ID
           scopes (list): List of required permission scopes
   
       Returns:
           str: The access token, or None if authentication fails
       """
       app = PublicClientApplication(
           client_id,
           authority=f"https://login.microsoftonline.com/{tenant_id}"
       )
   
       print("Opening browser for interactive authentication...")
       token_response = app.acquire_token_interactive(scopes=scopes)
   
       if "access_token" in token_response:
           print("Successfully authenticated")
           return token_response["access_token"]
       else:
           print(f"Authentication failed: {token_response.get('error_description', 'Unknown error')}")
           return None
   
   # Uncomment the lines below to use interactive authentication
   token = get_access_token(tenant_id, client_id, scopes)
   print("Access token acquired via interactive login")
   

2. Esegui la cella del notebook. Verrà visualizzato il token Microsoft Entra restituito.

   

Informazioni sugli ambiti Code.* per l'API Livy

Quando i processi Spark vengono eseguiti tramite l'API Livy, gli Code.* ambiti controllano quali servizi esterni il runtime Spark può accedere per conto dell'utente autenticato. Due sono obbligatori; il resto è facoltativo a seconda del carico di lavoro.

Ambiti di Codice obbligatorio.*

Ambiti facoltativi code.*

Aggiungere questi ambiti solo se i processi Spark devono accedere ai servizi di Azure corrispondenti in fase di esecuzione.

Creare una sessione Spark dell'API Livy

1. Aggiungere un'altra cella del notebook e inserire questo codice.

   import json
   import requests
   
   api_base_url = "https://api.fabric.microsoft.com/"  # Base URL for Fabric APIs
   
   # Fabric Resource IDs - Replace with your workspace and lakehouse IDs
   workspace_id = "Fabric_WorkspaceID"
   lakehouse_id = "Fabric_LakehouseID"
   
   # Construct the Livy API session URL
   # URL pattern: {base_url}/v1/workspaces/{workspace_id}/lakehouses/{lakehouse_id}/livyapi/versions/{api_version}/sessions
   livy_api_session_url = (f"{api_base_url}v1/workspaces/{workspace_id}/lakehouses/{lakehouse_id}/"
                          f"livyapi/versions/2023-12-01/sessions")
   
   # Set up authentication headers
   headers = {"Authorization": f"Bearer {token}"}
   
   print(f"Livy API URL: {livy_api_session_url}")
   print("Creating Livy session...")
   
   try:
       # Create a new Livy session with default configuration
       create_livy_session = requests.post(livy_api_session_url, headers=headers, json={})
   
       # Check if the request was successful
       if create_livy_session.status_code == 202:
           session_info = create_livy_session.json()
           print('Livy session creation request submitted successfully')
           print(f'Session Info: {json.dumps(session_info, indent=2)}')
   
           # Extract session ID for future operations
           livy_session_id = session_info['id']
           livy_session_url = f"{livy_api_session_url}/{livy_session_id}"
   
           print(f"Session ID: {livy_session_id}")
           print(f"Session URL: {livy_session_url}")
   
       else:
           print(f"Failed to create session. Status code: {create_livy_session.status_code}")
           print(f"Response: {create_livy_session.text}")
   
   except requests.exceptions.RequestException as e:
       print(f"Network error occurred: {e}")
   except json.JSONDecodeError as e:
       print(f"JSON decode error: {e}")
       print(f"Response text: {create_livy_session.text}")
   except Exception as e:
       print(f"Unexpected error: {e}")
   

2. Esegui la cella del notebook e vedrai una riga stampata mentre viene creata la sessione Livy.

   

3. È possibile verificare che la sessione Livy venga creata usando [Visualizzare i processi nell'hub di monitoraggio](#View i processi nell'hub di monitoraggio).

Integrazione con ambienti Fabric

Per impostazione predefinita, questa sessione dell'API Livy viene eseguita nel pool di avvio predefinito per l'area di lavoro. In alternativa, è possibile usare Fabric Ambienti Creare, configurare e usare un ambiente in Microsoft Fabric per personalizzare il pool di Spark usato dalla sessione api Livy per questi processi Spark. Per usare un ambiente Fabric, aggiornare la cella precedente del notebook con questo payload JSON.

create_livy_session = requests.post(livy_base_url, headers = headers, json = {
    "conf" : {
        "spark.fabric.environmentDetails" : "{\"id\" : \""EnvironmentID""}"}
        }
)

Inviare un'istruzione spark.sql usando la sessione Spark dell'API Livy

1. Aggiungere un'altra cella del notebook e inserire questo codice.

       # call get session API
   import time
   
   table_name = "green_tripdata_2022"
   
   print("Checking session status...")
   
   # Get current session status
   get_session_response = requests.get(livy_session_url, headers=headers)
   session_status = get_session_response.json()
   print(f"Current session state: {session_status['state']}")
   
   # Wait for session to become idle (ready to accept statements)
   print("Waiting for session to become idle...")
   while session_status["state"] != "idle":
       print(f"   Session state: {session_status['state']} - waiting 5 seconds...")
       time.sleep(5)
       get_session_response = requests.get(livy_session_url, headers=headers)
       session_status = get_session_response.json()
   
   print("Session is now idle and ready to accept statements")
   
   # Execute a Spark SQL statement
   execute_statement_url = f"{livy_session_url}/statements"
   
   # Define your Spark SQL query - Replace with your actual table and query
   payload_data = {
       "code": "spark.sql(\"SELECT * FROM {table_name} WHERE column_name = 'some_value' LIMIT 10\").show()",
       "kind": "spark"  # Type of code (spark, pyspark, sql, etc.)
   }
   
   print("Submitting Spark SQL statement...")
   print(f"Query: {payload_data['code']}")
   
   try:
       # Submit the statement for execution
       execute_statement_response = requests.post(execute_statement_url, headers=headers, json=payload_data)
   
       if execute_statement_response.status_code == 200:
           statement_info = execute_statement_response.json()
           print('Statement submitted successfully')
           print(f"Statement Info: {json.dumps(statement_info, indent=2)}")
   
           # Get statement ID for monitoring
           statement_id = str(statement_info['id'])
           get_statement_url = f"{livy_session_url}/statements/{statement_id}"
   
           print(f"Statement ID: {statement_id}")
   
           # Monitor statement execution
           print("Monitoring statement execution...")
           get_statement_response = requests.get(get_statement_url, headers=headers)
           statement_status = get_statement_response.json()
   
           while statement_status["state"] != "available":
               print(f"   Statement state: {statement_status['state']} - waiting 5 seconds...")
               time.sleep(5)
               get_statement_response = requests.get(get_statement_url, headers=headers)
               statement_status = get_statement_response.json()
   
           # Retrieve and display results
           print("Statement execution completed!")
           if 'output' in statement_status and 'data' in statement_status['output']:
               results = statement_status['output']['data']['text/plain']
               print("Query Results:")
               print(results)
           else:
               print("No output data available")
   
       else:
           print(f"Failed to submit statement. Status code: {execute_statement_response.status_code}")
           print(f"Response: {execute_statement_response.text}")
   
   except Exception as e:
       print(f"Error executing statement: {e}")
   

2. Eseguire la cella del notebook. Dovresti vedere diverse righe incrementali stampate man mano che il lavoro viene inviato e i risultati vengono restituiti.

   

Inviare una seconda istruzione spark.sql usando la sessione Spark dell'API Livy

1. Aggiungere un'altra cella del notebook e inserire questo codice.

   print("Executing additional Spark SQL statement...")
   
   # Wait for session to be idle again
   get_session_response = requests.get(livy_session_url, headers=headers)
   session_status = get_session_response.json()
   
   while session_status["state"] != "idle":
       print(f"   Waiting for session to be idle... Current state: {session_status['state']}")
       time.sleep(5)
       get_session_response = requests.get(livy_session_url, headers=headers)
       session_status = get_session_response.json()
   
   # Execute another statement - Replace with your actual query
   payload_data = {
       "code": f"spark.sql(\"SELECT COUNT(*) as total_records FROM {table_name}\").show()",
       "kind": "spark"
   }
   
   print(f"Executing query: {payload_data['code']}")
   
   try:
       # Submit the second statement
       execute_statement_response = requests.post(execute_statement_url, headers=headers, json=payload_data)
   
       if execute_statement_response.status_code == 200:
           statement_info = execute_statement_response.json()
           print('Second statement submitted successfully')
   
           statement_id = str(statement_info['id'])
           get_statement_url = f"{livy_session_url}/statements/{statement_id}"
   
           # Monitor execution
           print("Monitoring statement execution...")
           get_statement_response = requests.get(get_statement_url, headers=headers)
           statement_status = get_statement_response.json()
   
           while statement_status["state"] != "available":
               print(f"   Statement state: {statement_status['state']} - waiting 5 seconds...")
               time.sleep(5)
               get_statement_response = requests.get(get_statement_url, headers=headers)
               statement_status = get_statement_response.json()
   
           # Display results
           print("Second statement execution completed!")
           if 'output' in statement_status and 'data' in statement_status['output']:
               results = statement_status['output']['data']['text/plain']
               print("Query Results:")
               print(results)
           else:
               print("No output data available")
   
       else:
           print(f"Failed to submit second statement. Status code: {execute_statement_response.status_code}")
   
   except Exception as e:
       print(f"Error executing second statement: {e}")
   

2. Eseguire la cella del notebook. Dovresti vedere diverse righe incrementali stampate man mano che il lavoro viene inviato e i risultati vengono restituiti.

   

Terminare la sessione Livy

1. Aggiungere un'altra cella del notebook e inserire questo codice.

   print("Cleaning up Livy session...")
   
   try:
       # Check current session status before deletion
       get_session_response = requests.get(livy_session_url, headers=headers)
       if get_session_response.status_code == 200:
           session_info = get_session_response.json()
           print(f"Session state before deletion: {session_info.get('state', 'unknown')}")
   
       print(f"Deleting session at: {livy_session_url}")
   
       # Delete the session
       delete_response = requests.delete(livy_session_url, headers=headers)
   
       if delete_response.status_code == 200:
           print("Session deleted successfully")
       elif delete_response.status_code == 404:
           print("Session was already deleted or not found")
       else:
           print(f"Delete request completed with status code: {delete_response.status_code}")
           print(f"Response: {delete_response.text}")
   
       print(f"Delete response details: {delete_response}")
   
   except requests.exceptions.RequestException as e:
       print(f"Network error during session deletion: {e}")
   except Exception as e:
       print(f"Error during session cleanup: {e}")
   

Visualizza i tuoi lavori nell'hub di monitoraggio

È possibile accedere all'hub di monitoraggio per visualizzare varie attività di Apache Spark selezionando Monitoraggio nei collegamenti di spostamento a sinistra.

1. Quando la sessione è in corso o in stato completato, è possibile visualizzare lo stato della sessione passando a Monitoraggio.

   

2. Selezionare e aprire il nome dell'attività più recente.

   

3. In questo caso di sessione dell'API Livy è possibile visualizzare gli invii di sessioni precedenti, i dettagli dell'esecuzione, le versioni di Spark e la configurazione. Osservare lo stato fermo in alto a destra.

   

Per riepilogare l'intero processo, è necessario un client remoto, ad esempio Visual Studio Code, un token Microsoft Entra app/SPN, l'URL dell'endpoint dell'API Livy, l'autenticazione per lakehouse e infine un'API Session Livy.

Contenuto correlato

• Documentazione dell'API REST Apache Livy
• Inviare processi batch Spark usando l'API Livy
• Informazioni generali sul monitoraggio di Apache Spark
• Dettagli dell'applicazione Apache Spark
• Supporto ad alta concorrenza nell'API Fabric Livy.