Partilhar via

Use a API do Livy para enviar e executar trabalhos em lote do Livy

Aplica-se a:✅ Engenharia de Dados e Ciência de Dados no Microsoft Fabric

Aprenda a enviar trabalhos em lote do Spark usando a API do Livy para Engenharia de Dados Fabric. A API Livy atualmente não suporta Azure Service Principal (SPN).

Pré-requisitos

A API Livy define um ponto de extremidade unificado para operações. Substitua os espaços reservados {Entra_TenantID}, {Entra_ClientID}, {Fabric_WorkspaceID} e {Fabric_LakehouseID} pelos valores apropriados ao seguir os exemplos deste artigo.

Configurar o código do Visual Studio para seu lote de API Livy

  1. Selecione Lakehouse Settings no seu Fabric Lakehouse.

    Captura de tela mostrando as configurações do Lakehouse.

  2. Navegue para a seção Livy endpoint.

    captura de tela mostrando o ponto de extremidade Lakehouse Livy e a cadeia de conexão de trabalho Session.

  3. Copie a cadeia de conexão da tarefa em lote (segunda caixa vermelha na imagem) para o seu código.

  4. Navegue até o Centro de administração do Microsoft Entra e copie a ID do aplicativo (cliente) e a ID do diretório (locatário) para o código.

    Captura de tela mostrando a visão geral do aplicativo Livy API no centro de administração do Microsoft Entra.

Crie um código Spark Batch e faça o upload para a sua Lakehouse

  1. Crie um .ipynb notebook no Visual Studio Code e insira o seguinte código

    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. Salve o arquivo Python localmente. Este pacote de código Python contém duas instruções Spark que funcionam em dados num Lakehouse e precisam ser carregadas para o seu Lakehouse. Você precisa do caminho ABFS da carga útil para referência no seu trabalho em lote da API Livy no Visual Studio Code, assim como do nome da tabela Lakehouse na instrução SQL Select.

    Captura de ecrã mostrando a célula de carga do Python.

  3. Carregue a carga útil do Python para a secção de ficheiros do seu Lakehouse. No explorador do Lakehouse, selecione Arquivos. Em seguida, selecione >Obter dados>Carregar arquivos. Selecione arquivos através do seletor de arquivos.

    Captura de ecrã mostrando a carga de dados na secção de ficheiros da Lakehouse.

  4. Após o ficheiro estar na secção de Ficheiros da sua Lakehouse, clique nos três pontos à direita do nome do ficheiro e selecione Propriedades.

    Captura de tela mostrando o caminho ABFS da carga útil nas Propriedades do arquivo no Lakehouse.

  5. Copie este caminho ABFS para a célula do Bloco de Anotações na etapa 1.

Autenticar uma sessão em lote do Livy API Spark usando um token de usuário do Microsoft Entra ou um token SPN do Microsoft Entra

Autenticar uma sessão em lote do Livy API Spark usando um token SPN do Microsoft Entra

  1. Crie um .ipynb notebook no Visual Studio Code e insira o código a seguir.

    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. Execute a célula do bloco de anotações, você verá o token Microsoft Entra retornado.

    Captura de tela mostrando o token SPN do Microsoft Entra retornado após a execução da célula.

Autenticar uma sessão do Livy API Spark usando um token de usuário do Microsoft Entra

  1. Crie um .ipynb notebook no Visual Studio Code e insira o código a seguir.

    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. Execute a célula do notebook, um pop-up deve aparecer no seu navegador permitindo que o utilizador escolha a identidade com a qual iniciar sessão.

    Captura de ecrã a mostrar o ecrã de início de sessão na aplicação Microsoft Entra.

  3. Depois de escolher a identidade com a qual entrar, você precisa aprovar as permissões da API de registro do aplicativo Microsoft Entra.

    Captura de tela mostrando as permissões da API do aplicativo Microsoft Entra.

  4. Feche a janela do navegador depois de concluir a autenticação.

    Captura de tela mostrando a autenticação concluída.

  5. No Visual Studio Code você deve ver o token do Microsoft Entra retornado.

    Captura de tela mostrando o token do Microsoft Entra retornado após executar a célula e fazer login.

Envie um Livy Batch e monitore o trabalho em lote.

  1. Adicione outra célula do bloco de notas e insira este código.

    # 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. Execute a célula do bloco de anotações, você verá várias linhas impressas à medida que o trabalho do Livy Batch é criado e executado.

    Captura de tela mostrando resultados no Visual Studio Code após o Livy Batch Job ter sido enviado com êxito.

  3. Para ver as alterações, navegue de volta para a sua Lakehouse.

Integração com ambientes de malha

Por padrão, essa sessão da API Livy é executada no pool inicial padrão para o espaço de trabalho. Como alternativa, você pode usar ambientes de malha Criar, configurar e usar um ambiente no Microsoft Fabric para personalizar o pool do Spark que a sessão da API do Livy usa para esses trabalhos do Spark. Para usar o seu Fabric Environment, atualize a célula anterior do caderno com esta alteração de uma linha.

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

Exibir seus trabalhos no hub de monitoramento

Você pode acessar o hub de monitoramento para visualizar várias atividades do Apache Spark selecionando Monitor nos links de navegação do lado esquerdo.

  1. Quando o trabalho em lote estiver concluído, você poderá exibir o status da sessão navegando até o Monitor.

    Captura de tela mostrando envios anteriores da API Livy no hub de monitoramento.

  2. Selecione e abra o nome da atividade mais recente.

    Captura de tela mostrando a atividade mais recente da API Livy no hub de monitoramento.

  3. Neste caso de sessão da API Livy, você pode ver o envio em lote anterior, os detalhes da execução, as versões do Spark e a configuração. Observe o status parado no canto superior direito.

    Captura de tela mostrando os detalhes mais recentes da atividade da API Livy no hub de monitoramento.

Para recapitular todo o processo, você precisa de um cliente remoto, como o Visual Studio Code, um token de aplicativo Microsoft Entra, o endereço da API Livy, autenticação no seu Lakehouse, um payload Spark no seu Lakehouse e, finalmente, uma sessão de API Livy para processamento em lote.

Conteúdos relacionados

  • Last updated on