使用 Livy API 提交和执行会话作业

适用于：✅Microsoft Fabric 中的数据工程和数据科学

了解如何使用 Livy API for Fabric 数据工程提交 Spark 会话作业。

先决条件

具有湖屋的 Fabric 高级版或试用版容量
具有 Jupyter Notebook、PySpark 和用于 Python 的 Microsoft 身份验证库（MSAL）的远程客户端（例如 Visual Studio Code）
或者是 Microsoft Entra 应用令牌。将应用程序注册到 Microsoft 标识平台
或者Microsoft Entra SPN 令牌。在 Microsoft Entra 中添加和管理应用程序凭据
对于湖屋中的一些数据，此示例使用纽约市出租车和豪华轿车委员会 green_tripdata_2022_08，这是加载到湖屋中的一个 parquet 文件

Livy API 定义用于操作的统一终结点。按照本文中的示例操作时，请将占位符 {Entra_TenantID}、{Entra_ClientID}、{Fabric_WorkspaceID}、{Fabric_LakehouseID} 替换为相应的值。

为 Livy API 会话配置 Visual Studio Code

在 Fabric 湖屋中选择“湖屋设置”。
导航到“Livy 终结点”部分。
请将图像中第一个红色框内的会话作业连接字符串复制到您的代码中。
导航到 Microsoft Entra 管理中心，并将应用程序（客户端）ID 和目录（租户）ID 复制到代码。

使用 Microsoft Entra 用户令牌或 Microsoft Entra SPN 令牌对 Livy API Spark 会话进行身份验证

使用 Microsoft Entra SPN 令牌对 Livy API Spark 会话进行身份验证

在 Visual Studio Code 中创建 .ipynb 笔记本并插入以下代码。

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)

运行笔记本单元格。你应该会看到返回的Microsoft Entra令牌。

使用 Microsoft Entra 用户令牌对 Livy API Spark 会话进行身份验证

在 Visual Studio Code 中创建 .ipynb 笔记本并插入以下代码。

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

运行笔记本单元格。您应该会看到返回的 Microsoft Entra 令牌。

创建 Livy API Spark 会话

添加另一个笔记本单元格并插入此代码。

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 == 200:
        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}")

运行笔记本单元格，应该会看到在创建 Livy 会话时打印了一行。
可以使用[在监视中心查看作业](#在监视中心查看作业)来验证是否已创建 Livy 会话。

与 Fabric 环境集成

默认情况下，此 Livy API 会话针对工作区的默认初学者池运行。或者，你可以使用 Fabric 环境。在 Microsoft Fabric 中创建、配置并使用环境来自定义由 Livy API 会话用于这些 Spark 作业的 Spark 池。若要使用 Fabric 环境，请使用此 json 有效负载更新以前的笔记本单元。

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

使用 Livy API Spark 会话提交 spark.sql 语句

添加另一个笔记本单元格并插入此代码。

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

运行笔记本单元格，你应该会看到在作业提交和结果返回时打印出的几条递增的行。

使用 Livy API Spark 会话提交第二个 spark.sql 语句

添加另一个笔记本单元格并插入此代码。

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

运行笔记本单元格，你应该会看到在作业提交和结果返回时打印出的几条递增的行。

结束 Livy 会话

添加另一个笔记本单元格并插入此代码。

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

在监视中心查看作业

可以通过在左侧导航链接中选择“监视”来访问监视中心，以查看各种 Apache Spark 活动。

当会话正在进行或处于完成状态后，可以通过导航到“监视”来查看会话状态。
选择并打开最新活动名称。
在此 Livy API 会话案例中，可以看到以前的会话提交、运行详细信息、Spark 版本和配置。请注意右上角的已停止状态。

若要回顾整个过程，需要一个远程客户端（例如 Visual Studio Code）、Microsoft Entra 应用/SPN 令牌、Livy API 终结点 URL、针对湖屋的身份验证以及会话 Livy API。

反馈

此页面是否有帮助？

Last updated on 2025-11-10

通过

使用 Livy API 提交和执行会话作业

先决条件

为 Livy API 会话配置 Visual Studio Code

使用 Microsoft Entra 用户令牌或 Microsoft Entra SPN 令牌对 Livy API Spark 会话进行身份验证

使用 Microsoft Entra SPN 令牌对 Livy API Spark 会话进行身份验证

使用 Microsoft Entra 用户令牌对 Livy API Spark 会话进行身份验证

创建 Livy API Spark 会话

与 Fabric 环境集成

使用 Livy API Spark 会话提交 spark.sql 语句

使用 Livy API Spark 会话提交第二个 spark.sql 语句

结束 Livy 会话

在监视中心查看作业

相关内容

反馈

其他资源