Usare l'API Livy per inviare ed eseguire processi batch Livy

Si applica a:✅ Ingegneria dei dati e scienza dei dati in Microsoft Fabric

Informazioni su come inviare processi batch Spark usando l'API Livy per Fabric Data Engineering. L'API Livy attualmente non supporta il Service Principal di Azure (SPN).

Prerequisiti

• Fabbrica Premium o capacità di prova con un lakehouse.

• Un client remoto, ad esempio Visual Studio Code con Jupyter Notebooks, PySpark e Microsoft Authentication Library (MSAL) per Python.

• Per accedere all'API REST di Fabric è necessario un token dell'app Microsoft Entra. Registrare un'applicazione con Microsoft Identity Platform.

• 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}e {Fabric_LakehouseID} con i valori appropriati quando si seguono gli esempi in questo articolo.

Configurare Visual Studio Code per l'API Livy Batch

1. Selezionare Impostazioni del Lakehouse nel tuo Fabric Lakehouse.

   

2. Passare alla sezione Livy endpoint.

   

3. Copia la stringa di connessione del processo batch (nella seconda casella rossa nell'immagine) nel tuo codice.

4. Passare al centro di amministrazione di Microsoft Entra e copiare sia l'ID dell'applicazione (client) che l'ID della directory (tenant) nel codice.

   

Creare un codice Spark Batch e caricarlo in Lakehouse

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

   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. Salvare il file Python in locale. Questo payload di codice Python contiene due istruzioni Spark che funzionano sui dati in un Lakehouse e devono essere caricate in Lakehouse. È necessario il percorso ABFS del payload per fare riferimento al processo batch dell'API Livy in Visual Studio Code e al nome della tabella Lakehouse nell'istruzione Select SQL.

   

3. Caricare il payload Python nella sezione dei file di Lakehouse. In Lakehouse Explorer selezionare File. Seleziona >Ottieni dati>Carica file. Selezionare i file tramite il selettore di file.

   

4. Dopo che il file si trova nella sezione File di Lakehouse, fare clic sui tre puntini a destra del nome file del payload e selezionare Proprietà.

   

5. Copiare questo percorso ABFS nella cella Notebook nel passaggio 1.

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

Autenticare una sessione batch dell'API Livy Spark usando un token SPN di Microsoft Entra

1. Creare un .ipynb notebook 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, dovresti vedere il token Microsoft Entra restituito.

   

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

1. Creare un .ipynb notebook 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 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. Esegui la cella del notebook, nel browser dovrebbe apparire una finestra popup che consente di scegliere l'identità con cui accedere.

   

3. Dopo aver scelto l'identità con cui eseguire l'accesso, è necessario approvare le autorizzazioni dell'API di registrazione dell'app Microsoft Entra.

   

4. Chiudere la finestra del browser dopo aver completato l'autenticazione.

   

5. In Visual Studio Code dovrebbe essere visualizzato il token Microsoft Entra restituito.

   

Invia un batch Livy e monitora il job batch.

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

   # 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. Eseguire la cella del notebook. Verranno visualizzate diverse righe stampate durante la creazione e l'esecuzione del processo Livy Batch.

   

3. Per visualizzare le modifiche, tornare a Lakehouse.

Integrazione con gli ambienti di 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 gli ambienti di infrastruttura 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 il tuo ambiente Fabric, aggiorna la cella del notebook precedente applicando questa modifica da una sola riga.

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

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 lo stato del processo batch è 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 l'invio in batch precedente, 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 dell'app Microsoft Entra, l'URL dell'endpoint dell'API Livy, l'autenticazione contro il tuo Lakehouse, un payload Spark nel tuo Lakehouse e infine una sessione batch dell'API Livy.

Contenuto correlato

• Documentazione dell'API REST Apache Livy
• Sottomettere lavori di sessione usando l'API Livy
• Informazioni generali sul monitoraggio di Apache Spark
• Dettagli dell'applicazione Apache Spark