Teilen über

Verwenden der Livy-API zum Übermitteln und Ausführen von Livy-Batchaufträgen

Gilt für:✅ Datentechnik und Data Science in Microsoft Fabric

Erfahren Sie, wie Sie Spark-Batchaufträge mithilfe der Livy-API für Fabric Data Engineering übermitteln. Die Livy-API unterstützt derzeit keinen Azure Service Principal (SPN).

Voraussetzungen

Die Livy-API definiert einen einheitlichen Endpunkt für Vorgänge. Ersetzen Sie die Platzhalter {Entra_TenantID}, {Entra_ClientID}, {Fabric_WorkspaceID} und {Fabric_LakehouseID} durch die entsprechenden Werte, wenn Sie den Beispielen in diesem Artikel folgen.

Konfigurieren von Visual Studio Code für Ihren Livy-API-Batch

  1. Wählen Sie Lakehouse-Einstellungen in Ihrem Fabric Lakehouse aus.

    Screenshot: Lakehouse-Einstellungen.

  2. Navigieren Sie zum Abschnitt Livy-Endpunkt.

    Screenshot: Lakehouse-Livy-Endpunkt und Verbindungszeichenfolge des Sitzungsauftrags

  3. Kopieren Sie die Verbindungszeichenfolge des Batchauftrags (zweites rotes Feld in der Abbildung) in Ihren Code.

  4. Navigieren Sie zum Microsoft Entra Admin Center, und kopieren Sie sowohl die Anwendungs-ID (Client-ID) als auch die Verzeichnis-ID (Mandant) in Ihren Code.

    Screenshot der Übersicht über die Livy-API-App im Microsoft Entra Admin Center.

Erstellen Sie einen Spark Batch-Code und laden Sie es in das Lakehouse hoch.

  1. Erstellen Sie ein .ipynb-Notebook in Visual Studio Code, und fügen Sie den folgenden Code ein

    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. Speichern Sie die Python-Datei lokal. Diese Python-Codenutzlast enthält zwei Spark-Anweisungen, die an Daten in einem Lakehouse arbeiten und in Ihr Lakehouse hochgeladen werden müssen. Sie benötigen den ABFS-Pfad der Nutzlast, um in Ihrem Livy-API-Batchauftrag in Visual Studio Code sowie in Ihrem Lakehouse-Tabellennamen in der Select-SQL-Anweisung darauf zu verweisen.

    Screenshot: Python-Nutzdatenzelle

  3. Laden Sie die Python-Nutzdaten in den Dateiabschnitt Ihres Lakehouse hoch. Wählen Sie im Lakehouse-Explorer "Dateien" aus. Wählen Sie >Data abrufen>Dateien hochladen aus. Wählen Sie Dateien über die Dateiauswahl aus.

    Screenshot: Nutzdaten im Abschnitt „Dateien“ des Lakehouse

  4. Nachdem sich die Datei im Abschnitt „Dateien“ Ihres Lakehouse befindet, klicken Sie rechts neben dem Dateinamen der Nutzlast auf die drei Punkte, und wählen Sie „Eigenschaften“ aus.

    Screenshot: Nutzdaten-ABFS-Pfad in den Eigenschaften der Datei im Lakehouse

  5. Kopieren Sie diesen ABFS-Pfad in Ihre Notebookzelle in Schritt 1.

Authentifizieren einer Livy-API Spark-Batchsitzung mit einem Microsoft Entra-Benutzertoken oder einem Microsoft Entra SPN-Token

Authentifizieren einer Livy-API Spark-Batchsitzung mit einem Microsoft Entra SPN-Token

  1. Erstellen Sie ein .ipynb-Notebook in Visual Studio Code, und fügen Sie den folgenden Code ein.

    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. Führen Sie die Notizbuchzelle aus, und das Microsoft Entra-Token sollte zurückgegeben werden.

    Screenshot des Microsoft Entra SPN-Tokens, das nach dem Ausführen der Zelle zurückgegeben wird.

Authentifizieren einer Livy-API Spark-Sitzung mit einem Microsoft Entra-Benutzertoken

  1. Erstellen Sie ein .ipynb-Notebook in Visual Studio Code, und fügen Sie den folgenden Code ein.

    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. Führen Sie das Notebookfenster aus. In Ihrem Browser sollte ein Popupfenster erscheinen, in dem Sie die Identität auswählen können, mit der Sie sich anmelden möchten.

    Screenshot des Anmeldebildschirms für die Microsoft Entra-App.

  3. Nachdem Sie die Identität für die Anmeldung ausgewählt haben, müssen Sie die Berechtigungen der Microsoft Entra-App-Registrierungs-API genehmigen.

    Screenshot der Microsoft Entra-App-API-Berechtigungen.

  4. Schließen Sie das Browserfenster nach Abschluss der Authentifizierung.

    Screenshot: Authentifizierung abgeschlossen.

  5. In Visual Studio Code sollten Sie das zurückgegebene Microsoft Entra-Token sehen.

    Screenshot: Microsoft Entra-Token, das nach dem Ausführen der Zelle und dem Anmelden zurückgegeben wurde

Übermitteln Sie einen Livy-Stapel und überwachen Sie den Stapelauftrag.

  1. Fügen Sie eine weitere Notebookzelle hinzu, und fügen Sie diesen Code ein.

    # 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. Führen Sie die Notebook-Zelle aus. Es sollten mehrere Zeilen ausgegeben werden, wenn der Livy Batch-Auftrag erstellt und ausgeführt wird.

    Screenshot mit Ergebnissen in Visual Studio Code, nachdem der Livy-Batchauftrag erfolgreich übermittelt wurde.

  3. Um alle Änderungen zu sehen, gehen Sie zurück zu Ihrem Lakehouse.

Integration in Fabric-Umgebungen

Diese Livy-API-Sitzung wird standardmäßig für den Standardstartpool für den Arbeitsbereich ausgeführt. Alternativ können Sie Fabric-Umgebungen zum Erstellen, Konfigurieren und Verwenden einer Umgebung in Microsoft Fabric verwenden, um den Spark-Pool anzupassen, den die Livy-API-Sitzung für diese Spark-Aufträge verwendet. Um Ihre Fabric-Umgebung zu verwenden, aktualisieren Sie die vorherige Notebook-Zelle mit dieser einzeiligen Änderung.

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

Anzeigen Ihrer Aufträge im Überwachungshub

Sie können auf den Überwachungshub zugreifen, um verschiedene Apache Spark-Aktivitäten anzuzeigen, indem Sie in den Navigationslinks auf der linken Seite „Überwachen“ auswählen.

  1. Wenn der Batchauftrag abgeschlossen ist, können Sie den Sitzungsstatus anzeigen, indem Sie zu Monitor navigieren.

    Screenshot früherer Livy-API-Übermittlungen im Überwachungshub.

  2. Wählen Sie den Namen der letzten Aktivität aus, und öffnen Sie sie.

    Screenshot der letzten Livy-API-Aktivität im Monitoring Hub.

  3. In diesem Livy-API-Sitzungsszenario können Sie Ihre vorherigen Batch-Einsendungen, Ausführungsdetails, Spark-Versionen und Konfigurationen einsehen. Beachten Sie den Status „Angehalten“ oben rechts.

    Screenshot: Details der letzten Livy-API-Aktivität im Überwachungshub

Um den gesamten Prozess zusammenzufassen, benötigen Sie einen Remote-Client wie Visual Studio Code, ein Microsoft Entra-App-Token, eine Livy-API-Endpunkt-URL, die Authentifizierung für den Zugriff auf Ihr Lakehouse, eine Spark-Payload in Ihrem Lakehouse und schließlich eine Batch-Livy-API-Sitzung.

Zugehöriger Inhalt

  • Last updated on