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

Si applica a:✅ Fabric Data Engineering and Data Science

Informazioni su come inviare processi batch Spark usando l'API Livy per Fabric Data Engineering. L'API Livy attualmente non supporta Azure principale del servizio (SPN).

Prerequisiti

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

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

• Per accedere all'API REST Fabric è necessario un token dell'app Microsoft Entra. Registrare un'applicazione con il 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 Lakehouse Settings nel tuo Fabric Lakehouse.

   

2. Passare alla sezione Livy endpoint.

   

3. Copiare la stringa di connessione del processo batch (seconda 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.

   

Creare un codice Spark Batch e caricarlo in Lakehouse

1. Creare un notebook .ipynb 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 una Lakehouse e devono essere caricate in Lakehouse. È necessario il percorso ABFS (Azure Blob File System) del payload per fare riferimento al processo batch dell'API Livy in Visual Studio Code e al nome della tabella Lakehouse nell'istruzione SQL SELECT.

   

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, selezionare i tre puntini (puntini di sospensione) 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 Microsoft Entra

Autenticare una sessione batch dell'API Livy Spark 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. Eseguire la cella del notebook e 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 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 di 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.

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 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 l'ambiente Fabric, aggiornare la cella del notebook precedente con questa modifica in 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 su Lakehouse, un payload Spark nella lakehouse e infine una sessione api Livy batch.

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