Uso de Livy API para enviar y ejecutar trabajos por lotes de Livy

Se aplica a:✅ Fabric Data Engineering and Data Science

Aprenda a enviar trabajos de Spark por lotes mediante la API de Livy para Fabric de ingeniería de datos. Actualmente, la API de Livy no admite el Principal de Servicio de Azure (SPN).

Requisitos previos

• Fabric Premium o Capacidad con un Lakehouse.

• Un cliente remoto, como Visual Studio Code con Jupyter Notebooks, PySpark y el Microsoft Authentication Library (MSAL) para Python.

• Se requiere un token de aplicación Microsoft Entra para acceder a la API rest de Fabric. Registrar una aplicación en la plataforma de identidad de Microsoft.

• Algunos datos de su instancia de almacén de lago, en este ejemplo se usa NYC Taxi & Limousine Commission green_tripdata_2022_08 un archivo parquet cargado en el lago.

Livy API define un punto de conexión unificado para las operaciones. Reemplace los marcadores de posición {Entra_TenantID}, {Entra_ClientID}, {Fabric_WorkspaceID}y {Fabric_LakehouseID} por los valores adecuados al seguir los ejemplos de este artículo.

Configuración de Visual Studio Code para Livy API Batch

1. Seleccione Lakehouse Settings en su Fabric Lakehouse.

   

2. Vaya a la sección punto de conexión de Livy.

   

3. Copie la cadena de conexión del trabajo por lotes (segundo cuadro rojo de la imagen) a su código.

4. Vaya a Microsoft Entra admin center y copie tanto el identificador de aplicación (cliente) como el identificador de directorio (inquilino) en el código.

   

Crea un código de Spark Batch y cárgalo en tu Lakehouse

1. Cree un cuaderno .ipynb en Visual Studio Code e inserte el código siguiente

   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. Guarde el archivo Python localmente. Esta carga de código Python contiene dos instrucciones Spark que procesan datos en un Lakehouse y deben subirse a tu Lakehouse. Necesita la ruta de ABFS (Azure Blob File System) de la carga útil para referenciar en su trabajo por lotes de la API de Livy en Visual Studio Code, así como el nombre de su tabla de Lakehouse en la instrucción SQL SELECT.

   

3. Cargue el script de Python en la sección de archivos de tu Lakehouse. En el explorador de Lakehouse, seleccione Archivos. A continuación, seleccione >Obtener datos>Cargar archivos. Seleccione los archivos a través del selector de archivos.

   

4. Una vez que el archivo se encuentra en la sección Archivos de Lakehouse, seleccione los tres puntos (puntos suspensivos) a la derecha del nombre de archivo de carga y seleccione Propiedades.

   

5. Copie esta ruta de acceso de ABFS a la celda de Notebook en el paso 1.

Autenticación de una sesión por lotes de Spark de Livy API mediante un token de usuario de Microsoft Entra o un token de SPN de Microsoft Entra

Autenticación de una sesión por lotes de Spark de Livy API mediante un token de SPN de Microsoft Entra

1. Cree un cuaderno .ipynb en Visual Studio Code e inserte el código siguiente.

   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. Ejecute la celda del cuaderno; debería ver que se devuelve el token de Microsoft Entra.

   

Autenticación de una sesión de Spark de livy API mediante un token de usuario de Microsoft Entra

1. Cree un cuaderno .ipynb en Visual Studio Code e inserte el código siguiente.

   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. Ejecute la celda del cuaderno; debería aparecer un elemento emergente en el explorador, lo que le permite elegir la identidad con la que iniciar sesión.

   

3. Después de elegir la identidad con la que iniciar sesión, debe aprobar los permisos de api de registro de aplicaciones de Microsoft Entra.

   

4. Cierre la ventana del explorador después de completar la autenticación.

   

5. En Visual Studio Code, debería ver el token de Microsoft Entra devuelto.

   

Descripción de los ámbitos Code.* para livy API

Cuando los trabajos de Spark se ejecutan a través de la API de Livy, los Code.* ámbitos controlan a qué servicios externos puede acceder Spark Runtime en representación del usuario autenticado. Se requieren dos; El resto es opcional en función de la carga de trabajo.

Ámbitos de Code.* requeridos

Ámbitos opcionales de Code.*

Agregue estos ámbitos solo si los trabajos de Spark necesitan acceder a los servicios de Azure correspondientes en tiempo de ejecución.

Envía una tarea de Livy y supervisa el trabajo por lotes.

1. Agregue otra celda del cuaderno e inserte este código.

   # 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. Ejecute la celda del cuaderno; verá varias líneas impresas a medida que se crea y ejecuta el trabajo por lotes de Livy.

   

3. Para ver los cambios, navegue de regreso a Lakehouse.

Integración con entornos de Fabric

De forma predeterminada, esta sesión de Livy API se ejecuta en el grupo de inicio predeterminado para el área de trabajo. También puede usar entornos de Fabric Crear, configurar y usar un entorno en Microsoft Fabric para personalizar el grupo de Spark que usa la sesión de la API de Livy para estos trabajos de Spark. Para usar el entorno de Fabric, actualice la celda anterior del cuaderno con este único cambio de línea.

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

Visualización de los trabajos en el centro de supervisión

Puede acceder al centro de supervisión para ver varias actividades de Apache Spark al seleccionar Supervisar en los vínculos de navegación del lado izquierdo.

1. Cuando el trabajo por lotes esté en estado completado, puede ver el estado de la sesión yendo a Monitor.

   

2. Seleccione y abra el nombre de la actividad más reciente.

   

3. En este caso de la sesión de la API de Livy, puede ver el envío previo por lotes, los detalles de la ejecución, las versiones de Spark y la configuración. Observe el estado detenido en la parte superior derecha.

   

Para resumir todo el proceso, necesita un cliente remoto, como Visual Studio Code, un token de aplicación de Microsoft Entra, la URL del punto de conexión de la Livy API, la autenticación contra tu Lakehouse, una carga útil de Spark en tu Lakehouse y, por último, una sesión por lotes de la Livy API.

Contenido relacionado

• Documentación de la API REST de Apache Livy
• Enviar trabajos de sesión mediante Livy API
• Resumen de supervisión de Apache Spark
• Detalles de la aplicación Apache Spark