Utiliser l’API Livy pour envoyer et exécuter des travaux de session

S'applique à :✅ Fabric Data Engineering and Data Science

Découvrez comment envoyer des travaux de session Spark à l’aide de l’API Livy pour Fabric Data Engineering.

Prérequis

• Fabric Premium ou Capacité d'essai avec un Lakehouse

• Un client distant tel que Visual Studio Code avec Jupyter Notebooks, PySpark et Microsoft Authentication Library (MSAL) pour Python

• Soit un jeton d'application Microsoft Entra. Registrer une application avec le Microsoft identity platform

• Ou un jeton Microsoft Entra SPN. Ajoutez et gérez les informations d’identification de l’application dans Microsoft Entra

• Certaines données dans votre lakehouse. Cet exemple utilise le fichier Parquet NYC Taxi & Limousine Commission green_tripdata_2022_08 chargé dans le lakehouse

L’API Livy définit un point de terminaison unifié pour les opérations. Remplacez les espaces réservés {Entra_TenantID}, {Entra_ClientID}, {Fabric_WorkspaceID}, {Fabric_LakehouseID} par vos valeurs appropriées lorsque vous suivez les exemples de cet article.

Configurer Visual Studio Code pour votre session d’API Livy

1. Sélectionnez Lakehouse Settings dans votre Fabric Lakehouse.

   

2. Accédez à la section Point de terminaison Livy.

   

3. Copiez la chaîne de connexion de la tâche Session (première zone rouge de l’image) dans votre code.

4. Accédez à Microsoft Entra admin center et copiez l’ID d’application (client) et l’ID d’annuaire (locataire) dans votre code.

   

Authentifier une session Spark d’API Livy à l’aide d’un jeton utilisateur Microsoft Entra ou d’un jeton SPN Microsoft Entra

Authentifier une session Spark d’API Livy à l’aide d’un jeton SPN Microsoft Entra

1. Créez un bloc-notes .ipynb dans Visual Studio Code et insérez le code suivant.

   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. Exécutez la cellule du bloc-notes. Vous devriez voir le jeton Microsoft Entra retourné.

   

Authentifier une session Spark d’API Livy à l’aide d’un jeton utilisateur Microsoft Entra

1. Créez un bloc-notes .ipynb dans Visual Studio Code et insérez le code suivant.

   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. Exécutez la cellule du bloc-notes. Vous devriez voir le jeton Microsoft Entra retourné.

   

Compréhension des périmètres Code.* pour l’API Livy

Lorsque vos travaux Spark s’exécutent via l’API Livy, les Code.* étendues contrôlent les services externes auxquels le runtime Spark peut accéder pour le compte de l’utilisateur authentifié. Deux sont nécessaires ; le reste est facultatif en fonction de votre charge de travail.

Étendues des codes requis.*

Domaines de code optionnels.*

Ajoutez ces étendues uniquement si vos travaux Spark doivent accéder aux services de Azure correspondants au moment de l’exécution.

Créer une session Spark d’API Livy

1. Ajoutez une autre cellule de notebook et insérez ce code.

   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. Exécutez la cellule de notebook. Vous devez voir une ligne imprimée lorsque la session Livy est créée.

   

3. Vous pouvez vérifier que la session Livy est créée à l’aide de [Afficher vos travaux dans le hub de supervision](#Afficher vos travaux dans le hub de supervision).

Intégration avec des environnements Fabric

Par défaut, cette session d’API Livy s’exécute sur le pool de démarrage par défaut de l’espace de travail. Vous pouvez également utiliser Fabric environnements Create, configurer et utiliser un environnement dans Microsoft Fabric pour personnaliser le pool Spark utilisé par la session API Livy pour ces travaux Spark. Pour utiliser un environnement Fabric, mettez à jour la cellule de notebook précédente avec cette charge utile JSON.

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

Envoyer une instruction spark.sql à l’aide de la session Spark de l’API Livy

1. Ajoutez une autre cellule de notebook et insérez ce code.

       # 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. Exécutez la cellule de notebook. Vous devez voir plusieurs lignes incrémentielles imprimées à mesure que le travail est envoyé et les résultats retournés.

   

Envoyer une deuxième instruction spark.sql à l’aide de la session Spark de l’API Livy

1. Ajoutez une autre cellule de notebook et insérez ce code.

   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. Exécutez la cellule de notebook. Vous devez voir plusieurs lignes incrémentielles imprimées à mesure que le travail est envoyé et les résultats retournés.

   

Terminer la session Livy

1. Ajoutez une autre cellule de notebook et insérez ce code.

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

Afficher vos travaux dans le hub de supervision

Vous pouvez accéder au hub de supervision pour afficher diverses activités Apache Spark en sélectionnant Superviser dans les liens de navigation de gauche.

1. Lorsque la session est en cours ou à l’état terminé, vous pouvez afficher l’état de la session en accédant à Superviser.

   

2. Sélectionnez et ouvrez le nom de l’activité la plus récente.

   

3. Dans ce cas de session API Livy, vous pouvez voir vos envois de sessions précédents, les détails d’exécution, les versions Spark et la configuration. Notez l’état Arrêté en haut à droite.

   

Pour récapituler l’ensemble du processus, vous avez besoin d’un client distant tel que Visual Studio Code, d’un jeton d’application/SPN Microsoft Entra, l’URL du point de terminaison de l’API Livy, l’authentification à votre Lakehouse, et enfin une API Livy de session.

Contenu connexe

• Documentation de l’API REST Apache Livy
• Envoyer des travaux par lots Spark à l’aide de l’API Livy
• Vue d’ensemble de la supervision d'Apache Spark
• Détails sur les applications Apache Spark
• Prise en charge d'une haute concurrence dans l'API Fabric Livy.