Aracılığıyla paylaş

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

Şunlar için geçerlidir:✅ Microsoft Fabric'te Veri Madenciliği ve Veri Bilimi

Doku Veri Mühendisliği için Livy API'sini kullanarak Spark toplu işleri göndermeyi öğrenin. Livy API şu anda Azure Hizmet Sorumlusu'nu (SPN) desteklememektedir.

Önkoşullar

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

Livy API Batch'iniz için Visual Studio Code'ı yapılandırma

  1. Fabric Lakehouse'unuzda Lakehouse Ayarları'nı seçin.

    Lakehouse ayarlarını gösteren ekran görüntüsü.

  2. Livy uç noktası bölümüne gidin.

    Lakehouse Livy uç noktası ve

  3. Batch işi bağlantı dizesini (görüntüdeki ikinci kırmızı kutu) kodunuza kopyalayın.

  4. Microsoft Entra yönetim merkezine gidin ve hem Uygulama (istemci) kimliği hem de Dizin (kiracı) kimliğini kodunuzla kopyalayın.

    Microsoft Entra yönetim merkezinde Livy API uygulamasına genel bakışı gösteren ekran görüntüsü.

Spark Batch kodu oluşturma ve Lakehouse'unuza yükleme

  1. Visual Studio Code'da not .ipynb defteri oluşturma ve aşağıdaki kodu ekleme

    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. Python dosyasını yerel olarak kaydedin. Bu Python kod yükü, bir Lakehouse'daki veriler üzerinde çalışan ve Lakehouse'unuza yüklenmesi gereken iki Spark deyimi içerir. Visual Studio Code'daki Livy API toplu işleminizde başvurmak için yükün ABFS yoluna ve Select SQL deyimindeki Lakehouse tablonuzun adına ihtiyacınız vardır.

    Python yük hücresini gösteren ekran görüntüsü.

  3. Python yükünü Lakehouse'unuzun dosyalar bölümüne yükleyin. Lakehouse gezgininde Dosyalar'ı seçin. Ardından >Veri al>Dosyaları karşıya yükle'yi seçin. Dosya seçici aracılığıyla dosyaları seçin.

    Lakehouse'un Dosyalar bölümünde yükü gösteren ekran görüntüsü.

  4. Dosya Lakehouse'unuzun Dosyalar bölümünde olduğunda, yüklemiş olduğunuz dosya adının sağındaki üç noktaya tıklayın ve Özellikler'i seçin.

    Lakehouse'daki dosyanın Özellikleri bölümünde yük ABFS yolunu gösteren ekran görüntüsü.

  5. Bu ABFS yolunu 1. adımda Not Defteri hücrenize kopyalayın.

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

Microsoft Entra SPN belirtecini kullanarak Livy API Spark toplu oturumu kimliğini doğrulama

  1. Visual Studio Code'da 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)

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

    Hücre çalıştırıldıktan sonra döndürülen Microsoft Entra SPN belirtecini gösteren ekran görüntüsü.

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

  1. Visual Studio Code'da 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 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 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. Not defteri hücresini çalıştırın, tarayıcınızda oturum açabileceğiniz kimliği seçmenize olanak sağlayan bir açılır pencere görünmelidir.

    Microsoft Entra uygulamasında oturum açma ekranını gösteren ekran görüntüsü.

  3. Oturum açmak istediğiniz kimliği seçtikten sonra Microsoft Entra uygulama kayıt API'sinin izinlerini onaylamanız gerekir.

    Microsoft Entra uygulama API'si izinlerini gösteren ekran görüntüsü.

  4. Kimlik doğrulamasını tamamladıktan sonra tarayıcı penceresini kapatın.

    Kimlik doğrulamasının tamamlanmasını gösteren ekran görüntüsü.

  5. Visual Studio Code'da Microsoft Entra belirtecinin döndüğünü görmelisiniz.

    Hücre çalıştırılıp oturum açıldıktan sonra döndürülen Microsoft Entra belirtecini gösteren ekran görüntüsü.

Livy Batch gönderin ve toplu işlemi denetleyin.

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

    # 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. Not defteri hücresini çalıştırın; Livy Batch işi oluşturulup çalıştırılırken birkaç satırın yazdırıldığını görmelisiniz.

    Livy Batch İşi başarıyla gönderildikten sonra Visual Studio Code'daki sonuçları gösteren ekran görüntüsü.

  3. Değişiklikleri görmek için Lakehouse'unuza geri dönün.

Kumaş Ortamlarıyla Tümleştirme

Varsayılan olarak, bu Livy API oturumu çalışma alanı için varsayılan başlangıç havuzunda çalışır. Alternatif olarak, Fabric Environments: Livy API oturumunun bu Spark işleri için kullandığı Spark havuzunu özelleştirmek amacıyla Microsoft Fabric'te bir ortam oluşturabilir, yapılandırabilir ve kullanabilirsiniz. Fabric Ortamınızı kullanmak için, önceki not defteri hücresini tek bir satır değişikliğiyle güncelleyin.

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

İ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.

  1. Toplu iş tamamlanma haline geldiğinde, İzleyici'ye giderek oturum durumunu görüntüleyebilirsiniz.

    İzleme hub'ında önceki Livy API gönderimlerini gösteren ekran görüntüsü.

  2. En son etkinlik adını seçin ve açın.

    İzleme hub'ında en son Livy API etkinliğini gösteren ekran görüntüsü.

  3. Bu Livy API oturum örneğinde, önceki toplu gönderiminizi, ç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.

    İzleme hub'ında en son Livy API etkinlik ayrıntılarını gösteren ekran görüntüsü.

İşlemin tamamını özetlemek için Visual Studio Code gibi bir uzak istemciye, Microsoft Entra uygulama belirtecine, Livy API uç noktası URL'sine, Lakehouse'unuzda kimlik doğrulamasına, Lakehouse'unuzda Spark yüküne ve son olarak toplu Livy API oturumuna ihtiyacınız vardır.

İlgili içerik

  • Last updated on