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

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

Aprenda como submeter trabalhos lotados no Spark usando a API Livy para Engenharia de Dados Fabric. A API do Livy atualmente não suporta o Azure Service Principal (SPN).

Pré-requisitos

• Fabric Premium ou Capacidade de teste com um Lakehouse.

• Um cliente remoto como Visual Studio Code com Jupyter Notebooks, PySpark e o Biblioteca de Autenticação da Microsoft (MSAL) para Python.

• É necessário um token de aplicação Microsoft Entra para aceder à API Fabric Rest. Regista uma candidatura no plataforma de identidades da Microsoft.

• Alguns dados em sua casa do lago, este exemplo usa NYC Taxi & Limousine Commission green_tripdata_2022_08 um arquivo de parquet carregado na casa do lago.

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.

Configure o Visual Studio Code para o seu Batch da API Livy

1. Selecione Lakehouse Settings no seu Fabric Lakehouse.

   

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

   

3. Copia a cadeia de conexão do trabalho em lote (segunda caixa vermelha na imagem) para o seu código.

4. Navegue até centro de administração Microsoft Entra e copie tanto o ID da Aplicação (cliente) como o ID do Diretório (inquilino) para o seu código.

   

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

1. Crie um caderno .ipynb em 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. Guarda o ficheiro Python localmente. Este payload de código Python contém duas instruções Spark que funcionam sobre dados num Lakehouse e precisam de ser carregadas para o seu Lakehouse. Precisas do caminho ABFS (Azure Blob File System) do payload para referenciá-lo no teu trabalho batch da API Livy no Visual Studio Code e do nome da tabela Lakehouse na instrução SQL SELECT.

   

3. Carrega o payload em Python para a secção de ficheiros do teu Lakehouse. No explorador do Lakehouse, selecione Arquivos. Em seguida, selecione >Obter dados>Carregar arquivos. Selecione arquivos através do seletor de arquivos.

   

4. Depois de o ficheiro estar na secção Ficheiros do seu Lakehouse, selecione os três pontos (elipses) à direita do nome do ficheiro de carga e selecione Propriedades.

   

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

Autenticar uma sessão batch do Spark da API Livy usando um token de utilizador Microsoft Entra ou um token SPN da Microsoft Entra

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

1. Crie um caderno .ipynb em Visual Studio Code e insira o seguinte código.

   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 notebook, deverá surgir o token Microsoft Entra.

   

Autenticar uma sessão Spark da API Livy usando um token de utilizador Microsoft Entra

1. Crie um caderno .ipynb em Visual Studio Code e insira o seguinte código.

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

   

3. Depois de escolher a identidade para iniciar sessão, precisa de aprovar as permissões da API de registo da aplicação Microsoft Entra.

   

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

   

5. No Visual Studio Code, deverá ver o token Microsoft Entra retornado.

   

Compreender os escopos do Code.* para a API do Livy

Quando os seus trabalhos Spark são executados através da API do Livy, os Code.* scopes controlam quais serviços externos o Spark Runtime pode aceder em nome do utilizador autenticado. São necessárias duas; Os restantes são opcionais dependendo da tua carga de trabalho.

Códigos obrigatórios.*

Code Opcionais.* contextos

Adicione estes escopos apenas se os seus trabalhos Spark precisarem de aceder aos serviços correspondentes do Azure em tempo de execução.

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.

   

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

Integração com ambientes Fabric

Por padrão, essa sessão da API Livy é executada no pool inicial padrão para o espaço de trabalho. Alternativamente, podes usar Fabric Ambientes Criar, configurar e usar um ambiente em Microsoft Fabric para personalizar o pool Spark que a sessão da API do Livy usa para estes 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.

   

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

   

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.

   

Para resumir todo o processo, precisas de um cliente remoto como Visual Studio Code, um token de aplicação Microsoft Entra, URL do endpoint da API do Livy, autenticação contra o teu Lakehouse, um payload Spark no teu Lakehouse e, finalmente, uma sessão batch da API do Livy.

Conteúdos relacionados

• Documentação da API REST do Apache Livy
• Enviar trabalhos de sessão usando a API do Livy
• Visão geral do monitoramento do Apache Spark
• Detalhes da aplicação Apache Spark