Použití rozhraní Livy API k odesílání a spouštění úloh relací

Platí pro:✅ Fabric Data Engineering and Data Science

Zjistěte, jak odesílat úlohy relací Sparku pomocí rozhraní Livy API pro Fabric datového inženýrství.

Požadavky

Fabric Premium nebo Testovací kapacita s Lakehouse
Vzdálený klient, například Visual Studio Code s Jupyter Notebooks, PySpark a Identity a ověřování Microsoftu (MSAL) pro Python
Token aplikace Microsoft Entra. Registrujte aplikaci pomocí Microsoft identity platform
Nebo token hlavního názvu služby (SPN) Microsoft Entra. Přidat a spravovat přihlašovací údaje aplikace v Microsoft Entra
Některá data ve vašem datovém jezeře: v tomto příkladu se používá soubor parquet NYC Taxi & Limousine Commission green_tripdata_2022_08, který je načten do datového jezera.

Rozhraní Livy API definuje jednotný koncový bod pro operace. Zástupné symboly {Entra_TenantID}, {Entra_ClientID}, {Fabric_WorkspaceID}, {Fabric_LakehouseID} nahraďte příslušnými hodnotami, když budete postupovat podle příkladů v tomto článku.

Konfigurace Visual Studio Code pro relaci rozhraní Livy API

Ve vašem Fabric Lakehouse vyberte Nastavení Lakehouse.
Přejděte do části Livy endpoint.
Zkopírujte připojovací řetězec úlohy relace (první červené pole na obrázku) do svého kódu.
Přejděte na Centrum pro správu Microsoft Entra a zkopírujte ID aplikace (klienta) i ID adresáře (tenanta) do kódu.

Ověřte relaci Sparku pomocí Livy API pomocí uživatelského tokenu Microsoft Entra nebo tokenu hlavního názvu služby (SPN) Microsoft Entra.

Ověření relace Spark rozhraní API Livy pomocí tokenu SPN Microsoft Entra

V Visual Studio Code vytvořte poznámkový blok .ipynb a vložte následující kód.

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)

Spusťte buňku poznámkového bloku. Měl by se zobrazit vrácený token Microsoft Entra.

Ověření relace Sparku rozhraní Livy API pomocí tokenu uživatele Microsoft Entra

V Visual Studio Code vytvořte poznámkový blok .ipynb a vložte následující kód.

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

Spusťte buňku poznámkového bloku. Měl by se zobrazit vrácený token Microsoft Entra.

Pochopení rozsahů Code.* v rozhraní Livy API

Když vaše úlohy Sparku běží přes rozhraní Livy API, Code.* obory řídí, k jakým externím službám může modul Runtime Spark přistupovat jménem ověřeného uživatele. Jsou vyžadovány dva; zbytek je volitelný v závislosti na vaší úloze.

Požadované obory kódu.*

Scope	Description
`Code.AccessFabric.All`	Umožňuje získat přístupové tokeny pro Fabric. Vyžaduje se pro všechny operace rozhraní Livy API.
`Code.AccessStorage.All`	Umožňuje získat přístupové tokeny k úložišti OneLake a Azure. Je vyžadováno pro čtení a zápis dat v lakehousech.

Volitelné obory kódu.*

Tyto obory přidejte pouze v případě, že úlohy Sparku potřebují přístup k odpovídajícím službám Azure za běhu.

Scope	Description	Kdy ho použít
`Code.AccessAzureKeyvault.All`	Umožňuje získat přístupové tokeny k Azure Key Vault.	Kód Sparku načte tajné kódy, klíče nebo certifikáty z Azure Key Vault.
`Code.AccessAzureDataLake.All`	Umožňuje získat přístupové tokeny k Azure Data Lake Storage Gen1.	Váš kód Spark čte nebo zapisuje do účtů Azure Data Lake Storage Gen1.
`Code.AccessAzureDataExplorer.All`	Umožňuje získat přístupové tokeny k Azure Data Explorer (Kusto).	Kód Sparku se dotazuje na nebo ingestuje data do/z klastrů Azure Data Explorer.
`Code.AccessSQL.All`	Umožňuje získat přístupové tokeny k Azure SQL.	Kód Sparku se musí připojit k Azure SQL databázím.

Poznámka:

Rozsahy Lakehouse.Execute.All a Lakehouse.Read.All jsou také povinné, ale nejsou součástí Code.* rodiny. Udělí oprávnění ke spouštění operací v Fabric lakehouse a čtení metadat.

Vytvořte relaci Spark pomocí rozhraní Livy API

Návod

Pokud vaše úloha vyžaduje souběžné spouštění více příkazů Sparku, zvažte místo toho použití relací s vysokou souběžností . Relace HC poskytují nezávislé kontexty provádění, které běží paralelně, zatímco systém spravuje opětovné použití podkladových relací Livy.

Přidejte další buňku poznámkového bloku a vložte tento kód.

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

Spusťte buňku poznámkového bloku. Při vytváření relace Livy by se měl zobrazit jeden řádek.
Relaci Livy můžete ověřit použitím odkazu [Zobrazit úlohy v centru monitorování](#View your jobs in the Monitoring hub).

Integrace s prostředími Fabric

Ve výchozím nastavení se tato relace API Livy spouští proti výchozí skupině zdrojů pro pracovní prostor. Alternativně můžete použít Fabric Environments Create, konfigurovat a použít prostředí ve Fabric k přizpůsobení Spark poolu, který Livy API session používá pro tyto Spark úlohy. Pokud chcete použít prostředí Fabric, aktualizujte předchozí buňku poznámkového bloku pomocí tohoto datového formátu json.

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

Odeslat příkaz Spark SQL pomocí Spark relace v rozhraní Livy API

Přidejte další buňku poznámkového bloku a vložte tento kód.

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

Spusťte buňku notebooku. Měli byste vidět několik postupných řádků, které se zobrazí při odeslání úlohy a návratu výsledků.

Odeslat druhý příkaz spark.sql pomocí relace Spark v rozhraní Livy API

Přidejte další buňku poznámkového bloku a vložte tento kód.

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

Spusťte buňku notebooku. Měli byste vidět několik postupných řádků, které se zobrazí při odeslání úlohy a návratu výsledků.

Ukončení relace Livy

Přidejte další buňku poznámkového bloku a vložte tento kód.

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

Zobrazení úloh v centru monitorování

K centru monitorování se dostanete tak, že vyberete Možnost Monitorování v navigačních odkazech na levé straně a zobrazíte různé aktivity Apache Sparku.

Když relace probíhá nebo je dokončená, můžete stav relace zobrazit tak, že přejdete na Monitorování.
Vyberte a otevřete název poslední aktivity.
V tomto případě relace rozhraní Livy API uvidíte předchozí odeslání relací, podrobnosti o spuštění, verze Sparku a konfiguraci. Všimněte si stavu zastavení v pravém horním rohu.

K rekapitulaci celého procesu potřebujete vzdáleného klienta, jako je Visual Studio Code, token aplikace nebo hlavního názvu služby (SPN) Microsoft Entra, adresa URL koncového bodu rozhraní Livy API, ověření vůči vašemu Lakehouse a nakonec Session Livy API.

Dokumentace k rozhraní Apache Livy REST API
Odesílání dávkových úloh Sparku pomocí rozhraní Livy API
Přehled monitorování Apache Sparku
Podrobnosti o aplikaci Apache Spark
Vysoká podpora souběžnosti v rozhraní API Fabric Livy.

Váš názor

Byla tato stránka užitečná?

Last updated on 2026-04-10

Použití rozhraní Livy API k odesílání a spouštění úloh relací

Požadavky

Konfigurace Visual Studio Code pro relaci rozhraní Livy API

Ověřte relaci Sparku pomocí Livy API pomocí uživatelského tokenu Microsoft Entra nebo tokenu hlavního názvu služby (SPN) Microsoft Entra.

Ověření relace Spark rozhraní API Livy pomocí tokenu SPN Microsoft Entra

Ověření relace Sparku rozhraní Livy API pomocí tokenu uživatele Microsoft Entra

Pochopení rozsahů Code.* v rozhraní Livy API

Požadované obory kódu.*

Volitelné obory kódu.*

Vytvořte relaci Spark pomocí rozhraní Livy API

Integrace s prostředími Fabric

Odeslat příkaz Spark SQL pomocí Spark relace v rozhraní Livy API

Odeslat druhý příkaz spark.sql pomocí relace Spark v rozhraní Livy API

Ukončení relace Livy

Zobrazení úloh v centru monitorování

Související obsah

Váš názor

Další materiály