Share via

De Livy-API gebruiken om Livy-batchtaken te verzenden en uit te voeren

Van toepassing op:✅ Data-engineer ing en Datawetenschap in Microsoft Fabric

Meer informatie over het verzenden van Spark-batchtaken met behulp van de Livy-API voor Fabric Data Engineering. De Livy-API biedt momenteel geen ondersteuning voor Azure Service Principal (SPN).

Vereisten

De Livy-API definieert een uniform eindpunt voor bewerkingen. Vervang de tijdelijke aanduidingen {Entra_TenantID}, {Entra_ClientID}, {Fabric_WorkspaceID} en {Fabric_LakehouseID} door de juiste waarden wanneer u de voorbeelden in dit artikel volgt.

Visual Studio Code configureren voor uw Livy API Batch

  1. Selecteer Lakehouse-instellingen in uw Fabric Lakehouse.

    Schermopname van Lakehouse-instellingen.

  2. Navigeer naar de Livy-eindpuntsectie.

    Schermopname van het Lakehouse Livy-eindpunt en de sessietaakverbindingsreeks.

  3. Kopieer de Batch-taak verbindingsreeks (tweede rood vak in de afbeelding) naar uw code.

  4. Navigeer naar het Microsoft Entra-beheercentrum en kopieer zowel de toepassings-id (client-id) als de map-id (tenant) naar uw code.

    Schermopname van het overzicht van de Livy-API-app in het Microsoft Entra-beheercentrum.

Een Spark Batch-code maken en uploaden naar uw Lakehouse

  1. .ipynb Een notebook maken in Visual Studio Code en de volgende code invoegen

    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. Sla het Python-bestand lokaal op. Deze Python-code-payload bevat twee Spark-instructies die werken met gegevens in een Lakehouse en die moeten worden geüpload naar uw Lakehouse. U hebt het ABFS-pad van de payload nodig om naar uw Livy API-batchtaak in Visual Studio Code te verwijzen, en de naam van uw Lakehouse-tabel in de SELECT SQL-instructie.

    Schermopname van de Python payload-cel.

  3. Upload de Python-nettolading naar de bestandssectie van uw Lakehouse. Selecteer in de Lakehouse-verkenner Bestanden. Selecteer vervolgens >Gegevens ophalen>Bestanden uploaden. Selecteer bestanden via de bestandskiezer.

    Schermopname van payload in de sectie 'Bestanden' van Lakehouse.

  4. Nadat het bestand in de sectie Bestanden van uw Lakehouse staat, klikt u op de drie puntjes rechts van de bestandsnaam van uw payload en selecteert u Eigenschappen.

    Schermopname van het payload ABFS-pad in de eigenschappen van het bestand in de Lakehouse.

  5. Kopieer dit ABFS-pad naar uw notebookcel in stap 1.

Een Livy API Spark-batchsessie verifiëren met behulp van een Microsoft Entra-gebruikerstoken of een Microsoft Entra SPN-token

Een Livy API Spark-batchsessie verifiëren met behulp van een Microsoft Entra SPN-token

  1. Maak een .ipynb notebook in Visual Studio Code en voeg de volgende code in.

    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. Voer de notebookcel uit. U ziet dat het Microsoft Entra-token is geretourneerd.

    Schermopname van het Microsoft Entra SPN-token dat is geretourneerd nadat de cel is uitgevoerd.

Een Livy API Spark-sessie verifiëren met behulp van een Microsoft Entra-gebruikerstoken

  1. Maak een .ipynb notebook in Visual Studio Code en voeg de volgende code in.

    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. Voer de notebookcel uit. Er moet een pop-up worden weergegeven in uw browser, zodat u de identiteit kunt kiezen waarmee u zich kunt aanmelden.

    Schermopname van het aanmeldingsscherm voor de Microsoft Entra-app.

  3. Nadat u de identiteit hebt gekozen waarmee u zich wilt aanmelden, moet u de API-machtigingen voor de registratie-API voor Microsoft Entra-apps goedkeuren.

    Schermopname van API-machtigingen voor Microsoft Entra-apps.

  4. Sluit het browservenster nadat u de verificatie hebt voltooid.

    Schermopname van het voltooien van de verificatie.

  5. In Visual Studio Code zou je het Microsoft Entra-token moeten zien terugkomen.

    Screenshot van het Microsoft Entra-token dat is geretourneerd nadat de cel is uitgevoerd en u bent ingelogd.

Verzend een Livy Batch en bewaak batchtaak.

  1. Voeg nog een notebookcel toe en voeg deze code in.

    # 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. Voer de notebookcel uit en u zou verschillende lijnen tekst moeten zien die worden afgedrukt terwijl de Livy Batch-taak wordt gemaakt en uitgevoerd.

    Schermopname van de resultaten in Visual Studio Code nadat de Livy Batch-taak is verzonden.

  3. Als u de wijzigingen wilt zien, gaat u terug naar uw Lakehouse.

Integratie met Fabric-omgevingen

Deze Livy API-sessie wordt standaard uitgevoerd op basis van de standaardstartgroep voor de werkruimte. U kunt ook Infrastructuuromgevingen maken, configureren en gebruiken in Microsoft Fabric om de Spark-pool aan te passen die door de Livy API-sessie wordt gebruikt voor deze Spark-taken. Als u uw Fabric-omgeving wilt gebruiken, werkt u de vorige notebookcel bij met deze regelwijziging.

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

Uw taken weergeven in de Bewakingshub

U hebt toegang tot de Bewakingshub om verschillende Apache Spark-activiteiten weer te geven door Monitor te selecteren in de navigatiekoppelingen aan de linkerkant.

  1. Wanneer de batchtaak is voltooid, kunt u de sessiestatus bekijken door naar Monitor te navigeren.

    Schermopname van eerdere Livy API-inzendingen in de Monitoring Hub.

  2. Selecteer en open de naam van de meest recente activiteit.

    Schermopname van de meest recente Livy-API-activiteit in de Bewakingshub.

  3. In dit geval van livy-API-sessie kunt u uw vorige batchinzending zien, details uitvoeren, Spark-versies en configuratie. Let op de gestopte status rechtsboven.

    Schermopname met de meest recente details van de Livy-API-activiteit in de Bewakingshub.

Als u het hele proces wilt terugvatten, hebt u een externe client nodig, zoals Visual Studio Code, een Microsoft Entra-app-token, de URL van het Livy API-eindpunt, verificatie voor uw Lakehouse, een Spark-nettolading in uw Lakehouse en ten slotte een batch-Livy API-sessie.

Gerelateerde inhoud

  • Last updated on