Compartir a través de

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

Se aplica a:✅ Ingeniería de datos y ciencia de datos en Microsoft Fabric

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

Requisitos previos

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 la sesión por lotes de Livy API

  1. Seleccione Configuración de Lakehouse en Fabric Lakehouse.

    Captura de pantalla que muestra la configuración del Lakehouse.

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

    Recorte de pantalla en el que se muestra el punto de conexión del almacén de lago de datos de Livy y la cadena de conexión del trabajo de sesión.

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

  4. Vaya a Centro de administración de Microsoft Entra y copie el identificador de aplicación (cliente) y el identificador de directorio (inquilino) en el código.

    Recorte de pantalla que muestra información general de la aplicación Livy API en el centro de administración de Microsoft Entra.

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 siguiente código

    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 de Python localmente. Esta carga de código de Python contiene dos instrucciones Spark que funcionan en los datos de un almacén de lago de datos y deben cargarse en su almacén de lago de datos. Necesita la ruta de acceso ABFS de la carga para referenciar en el trabajo por lotes de la API de Livy en Visual Studio Code y el nombre de la tabla de Lakehouse en la instrucción SQL SELECT.

    Recorte de pantalla que muestra la celda de carga de Python.

  3. Cargue la carga útil de Python en la sección de archivos de su 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.

    Recorte de pantalla que muestra la carga en la sección Archivos del almacén de lago de datos.

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

  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 siguiente código.

    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.

    Captura de pantalla que muestra el token SPN de Microsoft Entra devuelto después de ejecutar la celda.

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 siguiente código.

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

    Recorte de pantalla que muestra la pantalla de inicio de sesión en la aplicación Microsoft Entra.

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

    Recorte de pantalla que muestra los permisos de la API de aplicaciones de Microsoft Entra.

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

    Recorte de pantalla que muestra la autenticación completada.

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

    Recorte de pantalla que muestra el token de Microsoft Entra devuelto después de ejecutar la celda y iniciar sesió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.

    Recorte de pantalla que muestra los resultados en Visual Studio Code después de enviar correctamente el trabajo de Livy Batch.

  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. Como alternativa, puede usar entornos de Tejido 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.

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

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

  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.

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

Para volver a resumir todo el proceso, necesita un cliente remoto, como Visual Studio Code, un token de aplicación de Microsoft Entra, la dirección URL del punto de conexión de Livy API, la autenticación en el almacén de lago de datos, una carga de Spark en el almacén de lago de datos y, por último, una sesión de Livy API por lotes.

Contenido relacionado

  • Last updated on