Utiliser l’API Livy pour envoyer et exécuter des travaux par lots Livy

S’applique à :✅ l’engineering et la science des données dans Microsoft Fabric

Découvrez comment envoyer des tâches de traitement par lots Spark à l’aide de l’API Livy pour l’ingénierie des données Fabric. Actuellement, l’API Livy ne prend pas en charge le principal de service Azure (SPN).

Prérequis

• Capacité Fabric Premium ou d’évaluation avec un lakehouse.

• Un client distant tel que Visual Studio Code avec Jupyter Notebooks, PySpark et la bibliothèque d’authentification Microsoft (MSAL) pour Python.

• Un jeton d’application Microsoft Entra est nécessaire pour accéder à l’API REST Fabric. Inscrire une application à l’aide de la plateforme d’identités Microsoft.

• 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} et {Fabric_LakehouseID} par vos valeurs appropriées lorsque vous suivez les exemples de cet article.

Configurer Visual Studio Code pour votre lot d’API Livy

1. Sélectionnez Paramètres Lakehouse dans votre lakehouse Fabric.

   

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

   

3. Copiez la chaîne de connexion du travail par lots (deuxième zone rouge dans l’image) dans votre code.

4. Accédez au centre d’administration Microsoft Entra et copiez l’ID d’application (client) et l’ID d’annuaire (locataire) dans votre code.

   

Créez un code Spark Batch et chargez-le dans votre Lakehouse

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

   import sys
   import os
   
   from pyspark.sql import SparkSession
   from pyspark.conf import SparkConf
   from pyspark.sql.functions import col
   
   if __name__ == "__main__":
   
       #Spark session builder
       spark_session = (SparkSession
           .builder
           .appName("batch_demo") 
           .getOrCreate())
   
       spark_context = spark_session.sparkContext
       spark_context.setLogLevel("DEBUG")  
   
       tableName = spark_context.getConf().get("spark.targetTable")
   
       if tableName is not None:
           print("tableName: " + str(tableName))
       else:
           print("tableName is None")
   
       df_valid_totalPrice = spark_session.sql("SELECT * FROM green_tripdata_2022 where total_amount > 0")
       df_valid_totalPrice_plus_year = df_valid_totalPrice.withColumn("transaction_year", col("lpep_pickup_datetime").substr(1, 4))
   
   
       deltaTablePath = f"Tables/{tableName}CleanedTransactions"
       df_valid_totalPrice_plus_year.write.mode('overwrite').format('delta').save(deltaTablePath)
   

2. Enregistrez le fichier Python localement. Cette charge utile de code Python contient deux instructions Spark qui fonctionnent sur des données dans un Lakehouse et qui doivent être chargées dans votre Lakehouse. Vous avez besoin du chemin ABFS de la charge utile pour référencer votre travail de lot d’API Livy dans Visual Studio Code et le nom de votre table Lakehouse dans l’instruction Select SQL.

   

3. Chargez la charge utile Python dans la section des fichiers de votre Lakehouse. Dans l’explorateur Lakehouse, sélectionnez Fichiers. > Sélectionnez Obtenir des données>Télécharger des fichiers. Sélectionnez des fichiers via le sélecteur de fichiers.

   

4. Une fois que le fichier est dans la section Fichiers de votre Lakehouse, cliquez sur les trois petits points à droite du nom de fichier de votre charge utile, puis sélectionnez Propriétés.

   

5. Copiez ce chemin ABFS dans votre cellule Notebook à l’étape 1.

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

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

1. Créez un notebook .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 affiché.

   

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

1. Créez un notebook .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 Microsoft Fabric API access
   scopes = [
       "https://api.fabric.microsoft.com/Lakehouse.Execute.All",      # Execute operations in lakehouses
       "https://api.fabric.microsoft.com/Lakehouse.Read.All",        # Read lakehouse metadata
       "https://api.fabric.microsoft.com/Item.ReadWrite.All",        # Read/write fabric items
       "https://api.fabric.microsoft.com/Workspace.ReadWrite.All",   # Access workspace operations
       "https://api.fabric.microsoft.com/Code.AccessStorage.All",    # Access storage from code
       "https://api.fabric.microsoft.com/Code.AccessAzureKeyvault.All",     # Access Azure Key Vault
       "https://api.fabric.microsoft.com/Code.AccessAzureDataExplorer.All", # Access Azure Data Explorer
       "https://api.fabric.microsoft.com/Code.AccessAzureDataLake.All",     # Access Azure Data Lake
       "https://api.fabric.microsoft.com/Code.AccessFabric.All"             # General Fabric access
   ]
   
   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 Azure Active Directory 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 de notebook. Une fenêtre contextuelle devrait apparaître dans votre navigateur pour vous permettre de choisir l’identité avec laquelle vous connecter.

   

3. Après avoir choisi l’identité à utiliser pour vous connecter, vous devez approuver les autorisations de l’API d’inscription d’application Microsoft Entra.

   

4. Fermez la fenêtre du navigateur après avoir procédé à l’authentification.

   

5. Dans Visual Studio Code, vous devriez voir le jeton Microsoft Entra retourné.

   

Soumettez un lot Livy et surveillez le travail de traitement par lots.

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

   # submit payload to existing batch session
   
   import requests
   import time
   import json
   
   api_base_url = "https://api.fabric.microsoft.com/v1"  # 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 Batch API URL
   # URL pattern: {base_url}/workspaces/{workspace_id}/lakehouses/{lakehouse_id}/livyApi/versions/{api_version}/batches
   livy_base_url = f"{api_base_url}/workspaces/{workspace_id}/lakehouses/{lakehouse_id}/livyApi/versions/2023-12-01/batches"
   
   # Set up authentication headers
   headers = {"Authorization": f"Bearer {token}"}
   
   print(f"Livy Batch API URL: {livy_base_url}")
   
   new_table_name = "TABLE_NAME"  # Name for the new table
   
   # Configure the batch job
   print("Configuring batch job parameters...")
   
   # Batch job configuration - Modify these values for your use case
   payload_data = {
       # Job name - will appear in the Fabric UI
       "name": f"livy_batch_demo_{new_table_name}",
   
       # Path to your Python file in the lakehouse
       "file": "<ABFSS_PATH_TO_YOUR_PYTHON_FILE>",  # Replace with your Python file path
   
       # Optional: Spark configuration parameters
       "conf": {
           "spark.targetTable": new_table_name,  # Custom configuration for your application
       },
   }
   
   print("Batch Job Configuration:")
   print(json.dumps(payload_data, indent=2))
   
   try:
       # Submit the batch job
       print("\nSubmitting batch job...")
       post_batch = requests.post(livy_base_url, headers=headers, json=payload_data)
   
       if post_batch.status_code == 202:
           batch_info = post_batch.json()
           print("Livy batch job submitted successfully!")
           print(f"Batch Job Info: {json.dumps(batch_info, indent=2)}")
   
           # Extract batch ID for monitoring
           batch_id = batch_info['id']
           livy_batch_get_url = f"{livy_base_url}/{batch_id}"
   
           print(f"\nBatch Job ID: {batch_id}")
           print(f"Monitoring URL: {livy_batch_get_url}")
   
       else:
           print(f"Failed to submit batch job. Status code: {post_batch.status_code}")
           print(f"Response: {post_batch.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: {post_batch.text}")
   except Exception as e:
       print(f"Unexpected error: {e}")
   

2. Exécutez la cellule de notebook. Vous devriez voir plusieurs lignes qui s’affichent lors de la création et de l’exécution du travail par lots Livy.

   

3. Pour voir les modifications, revenez à votre Lakehouse.

Intégration à 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 les environnements Fabric Créer, 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 votre environnement Fabric, mettez à jour la cellule de notebook précédente avec cette modification de ligne.

payload_data = {
    "name":"livybatchdemo_with"+ newlakehouseName,
    "file":"abfss://YourABFSPathToYourPayload.py", 
    "conf": {
        "spark.targetLakehouse": "Fabric_LakehouseID",
        "spark.fabric.environmentDetails" : "{\"id\" : \""EnvironmentID"\"}"  # remove this line to use starter pools instead of an environment, replace "EnvironmentID" with your environment ID
        }
    }

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 le travail par lots est à l’état terminé, vous pouvez voir 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 d’API Livy, vous pouvez voir votre envoi de lot précédent, 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 à distance tel que Visual Studio Code, d’un jeton d’application Microsoft Entra, de l’URL du point de terminaison de l’API Livy, de l’authentification auprès de votre Lakehouse, d’une charge de travail Spark dans votre Lakehouse et enfin d’une session d’API Livy par lots.

Contenu connexe

• Documentation de l’API REST Apache Livy
• Soumettre des travaux par session en utilisant l’API Livy
• Vue d’ensemble de la supervision d'Apache Spark
• Détails sur les applications Apache Spark