Oturum işlerini göndermek ve yürütmek için Livy API'sini kullanma

Şunlar için geçerlidir:✅ Fabric Veri Mühendisliği ve Veri Bilimi

Fabric Veri Mühendisliği için Livy API'sini kullanarak Spark oturum işlerini göndermeyi öğrenin.

Önkoşullar

Fabric Premium veya Trial capacity ile Lakehouse
Visual Studio Code gibi uzak bir istemci, Jupyter Notebooks, PySpark, ve Python için Microsoft Authentication Library (MSAL)
Ya bir Microsoft Entra uygulama belirteci. Microsoft kimlik platformunda bir uygulama kaydetme
Veya bir Microsoft Entra SPN belirteci. Microsoft Entra
Göl evinizdeki bazı veriler, bu örnekte NYC Taxi & Limousine Commission green_tripdata_2022_08 adlı parquet dosyası göl evine yüklendiğinde kullanılır.

Livy API,işlemler için birleşik bir uç nokta tanımlar. Bu makaledeki örnekleri izlediğinizde {Entra_TenantID}, {Entra_ClientID}, {Fabric_WorkspaceID}, {Fabric_LakehouseID} yer tutucularını uygun değerlerle değiştirin.

Livy API Oturumunuz için Visual Studio Code yapılandırma

Fabric Lakehouse'unuzda Lakehouse Settings öğesini seçin.
Livy uç noktası bölümüne gidin.

&;lt;c2><c1><c0><sb0>Lakehouse Livy endpoint ve oturum işi bağlantı dizesini gösteren ekran görüntüsü.</sb0></c0></c1></c2>
Oturum işinin bağlantı dizesini (görüntüdeki ilk kırmızı kutu) kodunuza kopyalayın.
Microsoft Entra yönetim merkezi adresine gidin ve hem Uygulama (istemci) Kimliği hem de Dizin (kiracı) kimliğini kodunuzla kopyalayın.

Microsoft Entra kullanıcı belirteci veya Microsoft Entra SPN belirteci kullanarak Livy API Spark oturumlarının kimliğini doğrulama

Microsoft Entra SPN belirteci kullanarak Livy API Spark oturumlarının kimliğini doğrulama

Visual Studio Code bir .ipynb not defteri oluşturun ve aşağıdaki kodu ekleyin.

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)

Notebook hücresini çalıştırın. Microsoft Entra belirtecinin döndürüldüğünü görmelisiniz.

Microsoft Entra kullanıcı belirteci kullanarak Livy API Spark oturumlarının kimliğini doğrulama

Visual Studio Code bir .ipynb not defteri oluşturun ve aşağıdaki kodu ekleyin.

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

Notebook hücresini çalıştırın. Microsoft Entra belirtecinin döndürüldüğünü görmelisiniz.

Livy API'sinin Code.* kapsamlarını anlama

Spark işleriniz Livy API aracılığıyla çalıştırıldığında kapsamlar, Code.* Spark Çalışma Zamanı'nın kimliği doğrulanmış kullanıcı adına hangi dış hizmetlere erişebileceğini denetler. İki gereklidir; geri kalanı, iş yükünüze bağlı olarak isteğe bağlıdır.

Gerekli Kod.* kapsamları

Scope	Açıklama
`Code.AccessFabric.All`	Fabric'e erişim tokenları almanızı sağlıyor. Tüm Livy API işlemleri için gereklidir.
`Code.AccessStorage.All`	OneLake ve Azure depolama alanına erişim belirteçleri almaya izin verir. Lakehouse'larda veri okuma ve yazma işlemleri için gereklidir.

İsteğe bağlı Code.* kapsamları

Bu kapsamları yalnızca Spark işlerinizin çalışma zamanında ilgili Azure hizmetlerine erişmesi gerekiyorsa ekleyin.

Scope	Açıklama	Ne zaman kullanılır?
`Code.AccessAzureKeyvault.All`	Azure Key Vault için erişim belirteçlerine erişim izni verir.	Spark kodunuz Azure Key Vault'tan anahtarları, gizli bilgileri veya sertifikaları alır.
`Code.AccessAzureDataLake.All`	1. Nesil Azure Data Lake Storage erişim belirteçlerinin alınabilmesini sağlar.	Spark kodunuz 1. Nesil Azure Data Lake Storage hesaplarından okur veya bu hesaplara yazar.
`Code.AccessAzureDataExplorer.All`	Azure Veri Gezgini (Kusto) için erişim belirteçleri almaya izin verir.	Spark kodunuz Azure Veri Gezgini kümelere veri sorgular veya veri alır.
`Code.AccessSQL.All`	Azure SQL erişim belirteçlerinin alınabilmesini sağlar.	Spark kodunuzun Azure SQL veritabanlarına bağlanması gerekir.

Uyarı

Lakehouse.Execute.All ve Lakehouse.Read.All kapsamları da gereklidir ancak Code.* ailesinin bir parçası değildir. Bunlar, sırasıyla Fabric lakehouse'larda işlem yürütme ve meta verileri okuma izni verir.

Bir Livy API Spark oturumu oluşturun

İpucu

İş yükünüz birden çok Spark deyiminin eşzamanlı olarak yürütülmesini gerektiriyorsa bunun yerine yüksek eşzamanlılık oturumları kullanmayı göz önünde bulundurun. HC oturumları, sistem temel alınan Livy oturumlarının yeniden kullanılmasını yönetirken paralel olarak çalışan bağımsız yürütme bağlamları sağlar.

Başka bir not defteri hücresi ekleyin ve bu kodu ekleyin.

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

Not defteri hücresini çalıştırın; Livy oturumu oluşturulduğunda tek bir satırın yazdırıldığını görmelisiniz.
Livy oturumunun oluşturulduğunu doğrulamak için [İzleme merkezinde işlerinizi kontrol etme](#View your jobs in the Monitoring hub) kullanabilirsiniz.

Fabric Ortamları ile Entegrasyon

Varsayılan olarak, bu Livy API oturumu çalışma alanı için varsayılan başlangıç havuzunda çalışır. Alternatif olarak, Fabric Environments Create, konfigürasyon ve Fabric'teki bir ortamı kullanarak Livy API oturumunun bu Spark işleri için kullandığı Spark havuzunu özelleştirebilirsiniz. Fabric Ortamını kullanmak için önceki not defteri hücresini bu JSON yüküyle güncelleyin.

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

Livy API Spark oturumunu kullanarak spark.sql deyimi gönderme

Başka bir not defteri hücresi ekleyin ve bu kodu ekleyin.

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

Not defteri hücresini çalıştırın, iş gönderildikçe ve sonuçlar döndürüldükçe birkaç artımlı satırın yazdırıldığını görmeniz gerekir.

Livy API Spark oturumunu kullanarak ikinci bir spark.sql deyimi gönderme

Başka bir not defteri hücresi ekleyin ve bu kodu ekleyin.

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

Not defteri hücresini çalıştırın, iş gönderildikçe ve sonuçlar döndürüldükçe birkaç artımlı satırın yazdırıldığını görmeniz gerekir.

Livy oturumunu sonlandırma

Başka bir not defteri hücresi ekleyin ve bu kodu ekleyin.

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

İzleme hub'ında işlerinizi görüntüleme

Sol taraftaki gezinti bağlantılarında İzleyici'yi seçerek çeşitli Apache Spark etkinliklerini görüntülemek için İzleme hub'ına erişebilirsiniz.

Oturum devam ederken veya tamamlandı durumundayken İzleyici'ye giderek oturum durumunu görüntüleyebilirsiniz.
En son etkinlik adını seçin ve açın.
Bu Livy API oturumu örneğinde, önceki oturum gönderimlerinizi, çalıştırma ayrıntılarınızı, Spark sürümlerini ve yapılandırmanızı görebilirsiniz. Sağ üst kısımdaki durduruldu durumuna dikkat edin.

İşlemin tamamını özetlemek için Visual Studio Code, Microsoft Entra uygulaması/SPN belirteci, Livy API uç nokta URL'si, Lakehouse'unuzda kimlik doğrulaması ve son olarak session Livy API'si gibi bir uzak istemci gerekir.

Apache Livy REST API belgeleri
Livy API'sini kullanarak Spark toplu işleri gönderme
Apache Spark izlemeye genel bakış
Apache Spark uygulama ayrıntısı
Fabric Livy API içinde yüksek eşzamanlılık desteği.

Geri Bildirim

Bu sayfayı yararlı buldunuz mu?

Son güncelleştirme: 2026-08-05

Oturum işlerini göndermek ve yürütmek için Livy API'sini kullanma

Önkoşullar

Livy API Oturumunuz için Visual Studio Code yapılandırma

Microsoft Entra kullanıcı belirteci veya Microsoft Entra SPN belirteci kullanarak Livy API Spark oturumlarının kimliğini doğrulama

Microsoft Entra SPN belirteci kullanarak Livy API Spark oturumlarının kimliğini doğrulama

Microsoft Entra kullanıcı belirteci kullanarak Livy API Spark oturumlarının kimliğini doğrulama

Livy API'sinin Code.* kapsamlarını anlama

Gerekli Kod.* kapsamları

İsteğe bağlı Code.* kapsamları

Bir Livy API Spark oturumu oluşturun

Fabric Ortamları ile Entegrasyon

Livy API Spark oturumunu kullanarak spark.sql deyimi gönderme

Livy API Spark oturumunu kullanarak ikinci bir spark.sql deyimi gönderme

Livy oturumunu sonlandırma

İzleme hub'ında işlerinizi görüntüleme

İlgili içerik

Geri Bildirim

Ek kaynaklar