Uso de Livy API para enviar y ejecutar trabajos de sesión

Se aplica a:✅ Fabric Data Engineering and Data Science

Obtenga información sobre cómo enviar trabajos de sesión de Spark mediante la API de Livy para la Ingeniería de Datos de Fabric.

Requisitos previos

• Fabric Premium o Capacidad de prueba con una instancia de Lakehouse

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

• Un token de aplicación Microsoft Entra. Registro de una aplicación en la plataforma de identidad de Microsoft

• O un token de SPN de Microsoft Entra. Agregar y administrar credenciales de aplicación en Microsoft Entra

• 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}, {Fabric_LakehouseID} por los valores adecuados cuando siga los ejemplos de este artículo.

Configuración de Visual Studio Code para la sesión de Livy API

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 de sesión (primer cuadro rojo de la imagen) en tu 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.

   

Autenticación de una sesión 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 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 el token de Microsoft Entra devuelto.

   

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 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. Ejecute la celda del cuaderno. 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 Livy API, los Code.* ámbitos controlan a qué servicios externos puede acceder Spark Runtime en nombre del usuario autenticado. Se requieren dos; El resto es opcional en función de la carga de trabajo.

Ámbitos requeridos de code.*

Á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.

Creación de una sesión de Spark de Livy API

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

   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. Ejecute la celda del cuaderno; debería ver una línea impresa a medida que se crea la sesión de Livy.

   

3. Puede comprobar que la sesión de Livy se crea mediante [Ver los trabajos en el centro de supervisión](#Ver los trabajos en el centro de supervisión).

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 un entorno de Fabric, actualice la celda del notebook anterior con esta carga JSON.

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

Envío de una instrucción spark.sql mediante la sesión de Spark de Livy API

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

       # 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. Ejecute la celda del cuaderno; debería ver varias líneas incrementales impresas a medida que se envía el trabajo y se devuelven los resultados.

   

Envío de una segunda instrucción spark.sql mediante la sesión de Spark de Livy API

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

   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. Ejecute la celda del cuaderno; debería ver varias líneas incrementales impresas a medida que se envía el trabajo y se devuelven los resultados.

   

Finalizar la sesión de Livy

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

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

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 la sesión está en curso o en estado completado, puede ver el estado de la sesión; para ello, vaya Supervisar.

   

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

   

3. En este caso de sesión de Livy API, puede ver los envíos de sesiones anteriores, los detalles de 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 Microsoft Entra app/SPN, la dirección URL del punto de conexión de Livy API, la autenticación en Lakehouse y, por último, una API de Livy de sesión.

Contenido relacionado

• Documentación de la API REST de Apache Livy
• Envío de trabajos por lotes de Spark mediante Livy API
• Resumen de supervisión de Apache Spark
• Detalles de la aplicación Apache Spark
• Compatibilidad con alta concurrencia en la API de Fabric Livy.