Použití rozhraní Livy API k odesílání a spouštění dávkových úloh Livy

Platí pro:✅ Fabric Data Engineering a Data Science

Zjistěte, jak odesílat dávkové úlohy pomocí rozhraní Livy API pro Fabric pro inženýrství dat. Rozhraní API Livy v současné době nepodporuje Azure Service Principal (SPN).

Požadavky

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

Konfigurace Visual Studio Code pro službu Livy API Batch

  1. V Fabric Lakehouse vyberte Nastavení Lakehouse.

    Snímek obrazovky s nastavením Lakehouse

  2. Přejděte do části Livy endpoint.

    snímek obrazovky zobrazující Lakehouse Livy koncový bod a připojovací řetězec k úloze Session.

  3. Zkopírujte úlohu Batch připojovací řetězec (druhé červené pole na obrázku) do kódu.

  4. Přejděte na Centrum pro správu Microsoft Entra a zkopírujte ID aplikace (klienta) i ID adresáře (tenanta) do kódu.

    Snímek obrazovky zobrazující přehled aplikace Livy API v Centru pro správu Microsoft Entra.

Vytvořte kód Spark Batch a nahrajte do Lakehouse

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

    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. Uložte Python soubor místně. Tento payload v Python kódu obsahuje dva Spark příkazy, které pracují s daty v lakehouse a je třeba je nahrát do vašeho lakehouse. Potřebujete cestu ABFS (Azure Blob File System) k payloadu, která odkazuje ve vašem Livy API batch jobu ve Visual Studio Code, a název vaší lakehouse tabulky v SQL příkazuSELECT.

    Snímek obrazovky znázorňující buňku payloadu Pythonu.

  3. Nahrajte Python payload do sekce Soubory vašeho Lakehouse. V Průzkumníku Lakehouse vyberte Soubory. Pak vyberte >Získat data>Nahrát soubory. Vyberte soubory pomocí nástroje pro výběr souborů.

    Snímek obrazovky znázorňující datovou část v části Soubory v Lakehouse

  4. Poté, co je soubor v sekci Soubory vašeho jezerního domu, vyberte tři tečky (elipsy) napravo od názvu souboru vašeho payload a vyberte Vlastnosti.

    Snímek obrazovky znázorňující cestu datové části ABFS ve vlastnostech souboru v Lakehouse

  5. Zkopírujte tuto ABFS cestu do buňky svého zápisníku v kroku 1.

Ověření dávkové relace rozhraní Livy API Sparku pomocí uživatelského tokenu Microsoft Entra nebo tokenu SPN Microsoft Entra

Autentizovat dávkovou relaci Sparku rozhraní API Livy pomocí tokenu SPN Microsoft Entra

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

  2. Spusťte buňku poznámkového bloku, měli byste vidět vrácený token Microsoft Entra.

    Screenshot zobrazující token hlavního názvu služby Microsoft Entra vrácený po spuštění buňky.

Autentizace relace Spark rozhraní Livy API pomocí uživatelského tokenu Microsoft Entra

  1. 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 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. Spusťte buňku poznámkového bloku, v prohlížeči by se mělo zobrazit vyskakovací okno, které vám umožní zvolit identitu pro přihlášení.

    Screenshot zobrazující přihlašovací obrazovku pro Microsoft Entra app.

  3. Jakmile zvolíte identitu pro přihlášení, musíte schválit oprávnění rozhraní API pro registraci Microsoft Entra aplikace.

    Snímek obrazovky zobrazující oprávnění API aplikace Microsoft Entra.

  4. Po dokončení ověřování zavřete okno prohlížeče.

    Snímek obrazovky znázorňující dokončení ověřování

  5. V Visual Studio Code byste měli vidět vrácený token Microsoft Entra.

    Screenshot zobrazující token Entra Microsoft vrácený po spuštění buňky a přihlášení.

Principy rozsahů Code.* pro Livy API

Když vaše úlohy Sparku běží přes rozhraní Livy API, obory řídí, Code.* 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 Sparku čte z účtů Azure Data Lake Storage Gen1 nebo do nich zapisuje.
Code.AccessAzureDataExplorer.All Umožňuje získat přístupové tokeny k Azure Data Explorer (Kusto). Kód Spark dotazuje nebo načítá data do/z clusterů 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í rodiny Code.*. Udělí oprávnění ke spouštění operací v Fabric lakehouses a k čtení metadat z Fabric lakehouses.

Odešlete Livy Batch a monitorujte dávkovou úlohu.

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

    # 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. Spusťte buňku poznámkového bloku, měli byste vidět několik řádků vytištěných v průběhu vytváření a spuštění úlohy Livy Batch.

    Screenshot zobrazující výsledky Visual Studio Code po úspěšném odeslání úlohy Batch Livy.

  3. Chcete-li vidět změny, vraťte se zpět do svého jezerního domu.

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. Chcete-li použít prostředí Fabric, aktualizujte předchozí buňku poznámkového bloku úpravou tohoto jednoho řádku.

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

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.

  1. Po dokončení dávkové úlohy můžete stav relace zobrazit tak, že přejdete na Monitorování.

    Snímek obrazovky znázorňující předchozí odeslání rozhraní Api Livy v centru monitorování

  2. Vyberte a otevřete název poslední aktivity.

    Snímek obrazovky zobrazující nejnovější aktivitu rozhraní Livy API v centru monitorování

  3. V tomto případě relace rozhraní Livy API uvidíte předchozí dávkové odeslání, podrobnosti o spuštění, verze Sparku a konfiguraci. Všimněte si stavu zastavení v pravém horním rohu.

    snímek obrazovky s nejnovějšími podrobnostmi o aktivitě rozhraní Livy API v centru monitorování

K rekapitulaci celého procesu potřebujete klienta na dálku, jako je Visual Studio Code, token aplikace Microsoft Entra, adresu URL koncového bodu rozhraní Livy API, ověření ve vašem systému lakehouse, datovou strukturu Spark v systému lakehouse a nakonec dávkovou relaci rozhraní Livy API.

Související obsah

  • Last updated on