Livy-ohjelmointirajapinnan avulla voit lähettää ja suorittaa Livy-erätöitä

Sovellettavissa: ✅ Fabric Data Engineering and Data Science

Opi, miten lähettää Spark-erätöitä Livy API:n avulla Fabric Data Engineeringissä. Livy API ei tällä hetkellä tue Azure Service Principalia (SPN).

Edellytykset

• Fabric Premium tai Kokeilukapasiteetti Lakehousella.

• Etäasiakas, kuten Visual Studio CodeJupyter Notebooks, PySpark ja Microsoft Authentication Library (MSAL) Python kanssa.

• Microsoft Entra -sovellustoken vaaditaan Fabric Rest API:n käyttämiseen. Rekisteröi hakemus Microsoftin käyttäjätietoympäristö:lle.

• Jotkin tiedot lakehousessa, tässä esimerkissä käytetään NEWC Taxi & Limousine Commission green_tripdata_2022_08 lakehouseen ladattua parquet-tiedostoa.

Livy-ohjelmointirajapinta määrittää toimintojen yhtenäisen päätepisteen. Korvaa paikkamerkit {Entra_TenantID}, {Entra_ClientID}, {Fabric_WorkspaceID} ja {Fabric_LakehouseID} sopivilla arvoilla, kun noudatat tämän artikkelin esimerkkejä.

Määritä Visual Studio Code Livy API Batchillesi

1. Valitse Lakehouse Settings Fabric Lakehousessasi.

   

2. Siirry Livy-päätepisteosioon.

   

3. Kopioi Batch job yhteysmerkkijono (kuvan toinen punainen laatikko) koodiisi.

4. Siirry Microsoft Entra -hallintakeskus ja kopioi sekä sovelluksen (asiakas) ID että Directory (tenant) ID koodiisi.

   

Luo Spark Batch -koodi ja lataa se Lakehouseen

1. Luo .ipynb-muistikirja Visual Studio Code:ssa ja lisää seuraava koodi

   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. Tallenna Python-tiedosto paikallisesti. Tämä Python-koodin hyötykuorma sisältää kaksi Spark-lausetta, jotka työskentelevät Lakehousen datalla ja jotka täytyy ladata Lakehouseen. Tarvitset hyötykuorman ABFS (Azure Blob File System) -polun, jotta voit viitata Livy API -erätehtävään Visual Studio Code:ssa ja Lakehouse-taulun nimeen SELECT SQL-lauseessa.

   

3. Lataa Python-hyötykuorma Lakehousesi tiedosto-osioon. Valitse Lakehousen resurssienhallinnassa Tiedostot. Valitse sitten >Nouda tiedot>Lataa tiedostot. Valitse tiedostot tiedostovalitsimen kautta.

   

4. Kun tiedosto on Lakehousesi Tiedostot-osiossa, valitse kolme pistettä (ellipsi) payload-tiedostonimen oikealta puolelta ja valitse Ominaisuudet.

   

5. Kopioi tämä ABFS-polku muistikirjan soluun vaiheessa 1.

Tunnista Livy API Spark -eräistunto joko Microsoft Entra -käyttäjätunnuksella tai Microsoft Entra SPN -tokenilla

Tunnista Livy API Spark -eräajoistunto Microsoft Entra SPN -tokenilla

1. Luo .ipynb-muistikirja Visual Studio Code:ssa ja lisää seuraava koodi.

   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. Suorita muistikirjan solu, niin Microsoft Entra -token palautuu.

   Kuvakaappaus, jossa Microsoft Entra SPN-tunniste palautui cell.

Livy API Spark -istunnon tunnistaminen Microsoft Entra -käyttäjätunnuksella

1. Luo .ipynb-muistikirja Visual Studio Code:ssa ja lisää seuraava koodi.

   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. Suorita muistikirjasolu. Selaimeen pitäisi ilmestyä ponnahdusikkuna, jonka avulla voit valita kirjautumisen käyttäjätiedot.

   

3. Kun olet valinnut kirjautumisen henkilöllisyyden, sinun täytyy hyväksyä Microsoft Entra -sovelluksen rekisteröinti-API-oikeudet.

   

4. Sulje selainikkuna todentamisen suorittamisen jälkeen.

   

5. Visual Studio Code:ssa Microsoft Entra -token palautetaan.

   

Understanding Code.* Scopes Livy API:lle

Kun Spark-työsi suoritetaan Livy API:n kautta, scopet Code.* ohjaavat, mitä ulkoisia palveluita Spark Runtime voi käyttää todennuksen saaneen käyttäjän puolesta. Kaksi vaaditaan; Loput ovat vapaaehtoisia riippuen työmäärästäsi.

Vaadittu koodi.* scopes

Valinnainen koodi.* kaukoputket

Lisää nämä scopet vain, jos Spark-tehtäväsi tarvitsevat pääsyn vastaaviin Azure-palveluihin ajonaikaisesti.

Lähetä Livy-erä ja valvo erätyötä.

1. Lisää toinen muistikirjasolu ja lisää tämä koodi.

   # 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. Suorita muistikirjasolu. Livy Batch -työn luomisen ja suorittamisen aikana pitäisi näkyä useita rivejä.

   

3. Jos haluat nähdä muutokset, siirry takaisin Lakehouseen.

Integraatio Fabric-ympäristöihin

Tämä Livy-ohjelmointirajapinnan istunto suoritetaan oletusarvoisesti työtilan oletusarvoista aloitussarjaa vasten. Vaihtoehtoisesti voit käyttää Fabric Environmentsia Luo, konfiguroi ja käytä ympäristöä Microsoft Fabric mukauttaaksesi Spark-poolia, jota Livy API -istunto käyttää näihin Spark-töihin. Käyttääksesi Fabric Environmentia, päivitä aiempi notebook-solu tällä yhdellä rivimuutoksella.

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

Tarkastele töitäsi valvontakeskuksessa

Voit käyttää valvontakeskusta tarkastelemaan erilaisia Apache Spark -toimintoja valitsemalla Vasemmanpuoleisten siirtymislinkkien Valvonta.

1. Kun erätyö on suoritettu, voit tarkastella istunnon tilaa siirtymällä kohtaan Valvonta.

   

2. Valitse ja avaa viimeisimmän toiminnon nimi.

   

3. Tässä Livy-ohjelmointirajapinta-istunnon tapauksessa voit tarkastella edellisen erän lähettämistä, suoritustietoja, Spark-versioita ja määritystä. Huomaa pysäytetty tila oikeassa yläkulmassa.

   

Koko prosessin tiivistämiseksi tarvitset etäasiakkaan kuten Visual Studio Code, Microsoft Entra-sovellustokenin, Livy API -päätepisteen URL-osoitteen, Lakehouse-tunnistautumisen, Spark-payloadin Lakehousessa ja lopuksi eräajoisen Livy API -istunnon.

Liittyvä sisältö

• Apache Livy REST -ohjelmointirajapinnan dokumentaatio
• Lähetä istuntotöitä Livy-ohjelmointirajapinnan avulla
• Apache Spark -valvonnan yleiskatsaus
• Apache Spark -sovelluksen tiedot