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

Se aplica a: Ingeniería de datos de tejido y ciencia de datos

Aprenda a enviar trabajos por lotes de Spark mediante Livy API for Fabric Data Engineering. Actualmente, la API Livy no admite Azure Principal de servicio (SPN).

Requisitos previos

Fabric Premium o Capacidad de prueba con un almacén de lago de datos.
Un cliente remoto, como Visual Studio Code con Jupyter Notebooks, PySpark y el Biblioteca de autenticación de Microsoft (MSAL) para Python.
Se requiere un token de aplicación Microsoft Entra para acceder a la API rest de Fabric. Registro de 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

Seleccione Configuración de Lakehouse en Fabric Lakehouse.

Captura de pantalla que muestra la configuración del Lakehouse.
Vaya a la sección Punto de conexión de Livy.
Copie la cadena de conexión de la tarea por lotes (segundo cuadro rojo de la imagen) a tu código.
Vaya al Centro de administración de Microsoft Entra y copie el ID de aplicación (cliente) y el ID de directorio (inquilino) en su código.

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

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)

Guarde el archivo Python localmente. Esta carga de código Python contiene dos instrucciones Spark que funcionan en los datos de tu Lakehouse y deben cargarse en tu Lakehouse. Necesita la ruta de acceso ABFS del paquete de datos para referenciar en su trabajo por lotes de la API Livy en Visual Studio Code y el nombre de su tabla Lakehouse en la instrucción SQL SELECT.
Cargue la carga útil de Python en la sección de archivos del Lakehouse. En el explorador de Lakehouse, seleccione Archivos. A continuación, seleccione Obtener datosCargar archivos. Seleccione los archivos a través del selector de archivos.

Recorte de pantalla que muestra la carga en la sección Archivos del almacén de lago de datos.
Después de que el archivo esté en la sección Archivos de su Lakehouse, haga clic en los tres puntos situados a la derecha del nombre de archivo de carga útil y seleccione Propiedades.

Recorte de pantalla que muestra la ruta de acceso ABFS de carga en las propiedades del archivo en el almacén de lago de datos.
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

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)

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

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

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.
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.
Cierre la ventana del explorador después de completar la autenticación.

Recorte de pantalla que muestra la autenticación completada.
En Visual Studio Code debería ver el token de Microsoft Entra devuelto.

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

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

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

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

Recorte de pantalla que muestra los envíos anteriores de Livy API en el centro de supervisión.
Seleccione y abra el nombre de la actividad más reciente.

Recorte de pantalla que muestra la actividad de Livy API más reciente en el centro de supervisión.
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.

Captura de pantalla que muestra los detalles de actividad de livy API más recientes en el centro de supervisión.

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

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

Comentarios

¿Le ha resultado útil esta página?

Last updated on 2026-03-05