Använda Livy-API:et för att skicka och köra sessionsjobb

Gäller för:✅ Fabric Data Engineering och Data Science

Lär dig hur du skickar Spark-sessionsjobb med livy-API:et för Fabric Data Engineering.

Förutsättningar

Livy-API:et definierar en enhetlig slutpunkt för åtgärder. Ersätt platshållarna {Entra_TenantID}, {Entra_ClientID}, {Fabric_WorkspaceID}, {Fabric_LakehouseID} med lämpliga värden när du följer exemplen i den här artikeln.

Konfigurera Visual Studio Code för din Livy API-session

  1. Välj Lakehouse Settings i din Fabric Lakehouse.

    Skärmbild som visar Lakehouse-inställningar.

  2. Gå till Livy-endpoint-avsnittet.

    screenshot som visar Lakehouse Livy-slutpunktens och sessionsjobbets anslutningssträng.

  3. Kopiera sessionsjobbs connection string (den röda rutan i bilden som visas först) till din kod.

  4. Gå till Microsoft Entra admin center och kopiera både program-ID:t (klient- och katalog-ID:t) till koden.

    Skärmbild som visar Översikt över Livy API-appen i Microsoft Entra admin center.

Autentisera en Livy API Spark-session med antingen en Microsoft Entra användartoken eller en Microsoft Entra SPN-token

Autentisera en Livy API Spark-session med en Microsoft Entra SPN-token

  1. Skapa en .ipynb notebook-fil i Visual Studio Code och infoga följande kod.

    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. Kör anteckningsboks-cellen. Du bör se att Microsoft Entra-token returneras.

    Skärmdump som visar en Microsoft Entra SPN-token som returneras efter att cellen har körts.

Autentisera en Livy API Spark-session med hjälp av en Microsoft Entra användartoken

  1. Skapa en .ipynb notebook-fil i Visual Studio Code och infoga följande kod.

    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. Kör anteckningsboks-cellen. Du bör se att Microsoft Entra-token returneras.

    Skärmdump som visar Microsoft Entra-användartoken som returnerades efter att cellen kördes.

Förstå Code.* omfånget för Livy-API:n

När dina Spark-jobb körs via Livy-API:et styr omfången Code.* vilka externa tjänster Spark Runtime kan komma åt för den autentiserade användarens räkning. Två krävs; resten är valfria beroende på din arbetsbelastning.

Nödvändig kod.* omfång

Scope Beskrivning
Code.AccessFabric.All Tillåter hämtning av åtkomsttoken till Microsoft Fabric. Krävs för alla Livy API-åtgärder.
Code.AccessStorage.All Tillåter att åtkomsttoken hämtas till OneLake och Azure lagring. Krävs för att läsa och skriva data i lakehouses.

Valfri kod.* omfång

Lägg bara till dessa omfång om dina Spark-jobb behöver komma åt motsvarande Azure tjänster vid körning.

Scope Beskrivning När det bör användas
Code.AccessAzureKeyvault.All Tillåter att åtkomsttoken hämtas till Azure Key Vault. Spark-koden hämtar hemligheter, nycklar eller certifikat från Azure Key Vault.
Code.AccessAzureDataLake.All Tillåter att åtkomsttoken hämtas till Azure Data Lake Storage Gen1. Spark-koden läser från eller skriver till Azure Data Lake Storage Gen1 konton.
Code.AccessAzureDataExplorer.All Tillåter att åtkomsttoken hämtas till Azure Data Explorer (Kusto). Spark-koden frågar eller matar in data till/från Azure Data Explorer kluster.
Code.AccessSQL.All Tillåter att åtkomsttoken hämtas till Azure SQL. Spark-koden måste ansluta till Azure SQL databaser.

Anmärkning

Omfången Lakehouse.Execute.All och Lakehouse.Read.All krävs också men ingår inte i Code.* familjen. De ger behörighet att köra åtgärder i och läsa metadata från Fabric lakehouses respektive.

Skapa en Livy API Spark-session

Tips/Råd

Om din arbetsbelastning kräver att flera Spark-instruktioner körs samtidigt bör du överväga att använda sessioner med hög samtidighet i stället. HC-sessioner ger oberoende körningskontexter som körs parallellt medan systemet hanterar återanvändning av underliggande Livy-sessioner.

  1. Lägg till ytterligare en notebook-cell och infoga den här koden.

    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. Kör notebook-cellen. Du bör se en rad som skrivs ut när Livy-sessionen skapas.

    Skärmbild som visar resultatet av den första notebook-cellkörningen.

  3. Du kan kontrollera att Livy-sessionen har skapats med hjälp av [Visa dina jobb i övervakningshubben](#View dina jobb i övervakningshubben).

Integrering med Fabric miljöer

Som standard körs den här Livy API-sessionen mot standardstartpoolen för arbetsytan. Du kan också använda Fabric Miljöer Skapa, konfigurera och använda en miljö i Microsoft Fabric för att anpassa Spark-poolen som Livy API-sessionen använder för dessa Spark-jobb. Om du vill använda en Fabric-miljö uppdaterar du den tidigare anteckningsbokscellen med den här JSON-payload.

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

Skicka en spark.sql-instruktion med Livy API Spark-sessionen

  1. Lägg till ytterligare en notebook-cell och infoga den här koden.

        # 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. Kör notebook-cellen. Du bör se flera rader skrivna ut stegvis när uppgiften skickas och resultaten returneras.

    Skärmbild som visar resultatet av den första notebook-cellen med Spark.sql körning.

Skicka en andra spark.sql-instruktion med Livy API Spark-sessionen

  1. Lägg till ytterligare en notebook-cell och infoga den här koden.

    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. Kör notebook-cellen. Du bör se flera rader skrivna ut stegvis när uppgiften skickas och resultaten returneras.

    Skärmbild som visar resultatet av den andra notebook-cellkörningen.

Avsluta Livy-sessionen

  1. Lägg till ytterligare en notebook-cell och infoga den här koden.

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

Visa dina jobb i övervakningshubben

Du kan komma åt övervakningshubben för att visa olika Apache Spark-aktiviteter genom att välja Övervaka i navigeringslänkarna till vänster.

  1. När sessionen pågår eller är i slutfört tillstånd kan du visa sessionsstatusen genom att gå till Övervaka.

    Skärmbild som visar tidigare Livy API-inlämningar i övervakningshubben.

  2. Välj och öppna det senaste aktivitetsnamnet.

    Skärmbild som visar den senaste Livy API-aktiviteten i övervakningshubben.

  3. I det här Livy API-sessionsfallet kan du se dina tidigare sessioner, körningsinformation, Spark-versioner och konfiguration. Observera den stoppade statusen längst upp till höger.

    Skärmbild som visar den senaste livy-API-aktivitetsinformationen i övervakningshubben.

För att sammanfatta hela processen behöver du en fjärrklient, till exempel Visual Studio Code, en Microsoft Entra app/SPN-token, Livy API-slutpunkts-URL, autentisering mot Lakehouse och slutligen ett Livy-API för session.

Relaterat innehåll

  • Last updated on