Χρήση του API Livy για την υποβολή και εκτέλεση εργασιών δέσμης livy

Ισχύει για:✅ Fabric Μηχανική Δεδομένων και Επιστήμη Δεδομένων

Μάθετε πώς μπορείτε να υποβάλετε ομαδικές εργασίες Spark χρησιμοποιώντας το Livy API for Fabric Data Engineering. Το Livy API προς το παρόν δεν υποστηρίζει το Azure Service Principal (SPN).

Προαπαιτούμενα στοιχεία

• Fabric Premium ή Δοκιμαστική χωρητικότητα με Lakehouse.

• Ένας απομακρυσμένος υπολογιστής-πελάτης, όπως Visual Studio Code με σημειωματάρια Jupyter, PySpark και Microsoft Authentication Library (MSAL) για Python.

• Απαιτείται ένα διακριτικό εφαρμογής Microsoft Entra για πρόσβαση στο Fabric Rest API. Καταχωρίστε μια αίτηση στο Πλατφόρμα ταυτοτήτων της Microsoft.

• Ορισμένα δεδομένα στο lakehouse σας, αυτό το παράδειγμα χρησιμοποιεί NYC Taxi & Limousine Commission green_tripdata_2022_08 ένα αρχείο parquet που φορτώνεται στο lakehouse.

Το API Livy ορίζει ένα ενοποιημένο τελικό σημείο για λειτουργίες. Αντικαταστήστε τα σύμβολα κράτησης θέσης {Entra_TenantID}, {Entra_ClientID}, {Fabric_WorkspaceID}και {Fabric_LakehouseID} με τις κατάλληλες τιμές σας, όταν ακολουθείτε τα παραδείγματα σε αυτό το άρθρο.

Διαμορφώστε τον κώδικα του Visual Studio Code για το Livy API Batch

1. Επιλέξτε Lakehouse Settings στο Fabric Lakehouse σας.

   

2. Μεταβείτε στην ενότητα τελικού σημείου Livy.

   

3. Αντιγράψτε το Batch job συμβολοσειρά σύνδεσης (δεύτερο κόκκινο πλαίσιο στην εικόνα) στον κώδικά σας.

4. Μεταβείτε στο Κέντρο διαχείρισης Microsoft Entra και αντιγράψτε τόσο το αναγνωριστικό εφαρμογής (πελάτη) όσο και το αναγνωριστικό καταλόγου (μισθωτή) στον κώδικά σας.

   

Δημιουργήστε έναν κωδικό δέσμης Spark και μεταφορτώστε τον στο Lakehouse σας

1. Δημιουργήστε ένα σημειωματάριο .ipynb στο Visual Studio Code και εισαγάγετε τον ακόλουθο κώδικα

   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 τοπικά. Αυτό το ωφέλιμο φορτίο κώδικα Python περιέχει δύο προτάσεις Spark που λειτουργούν σε δεδομένα σε ένα Lakehouse και πρέπει να αποσταλούν στο Lakehouse σας. Χρειάζεστε τη διαδρομή ABFS (Azure Blob File System) του ωφέλιμου φορτίου για αναφορά στην εργασία δέσμης Livy API στο Visual Studio Code και το όνομα του πίνακα Lakehouse στη δήλωση SELECT SQL.

   

3. Αποστείλετε το ωφέλιμο φορτίο Python στην ενότητα αρχείων του Lakehouse σας. Στην εξερεύνηση Lakehouse, επιλέξτε Αρχεία. Στη συνέχεια, επιλέξτε >Λήψη δεδομένων>Μεταφόρτωση αρχείων. Επιλέξτε αρχεία μέσω του επιλογέα αρχείων.

   

4. Αφού το αρχείο βρίσκεται στην ενότητα Αρχεία του Lakehouse σας, επιλέξτε τις τρεις τελείες (έλλειψη) στα δεξιά του ονόματος αρχείου ωφέλιμου φορτίου σας και επιλέξτε Ιδιότητες.

   

5. Αντιγράψτε αυτήν τη διαδρομή ABFS στο κελί του Σημειωματάριου στο βήμα 1.

Έλεγχος ταυτότητας μιας περιόδου λειτουργίας δέσμης Livy API Spark χρησιμοποιώντας είτε ένα διακριτικό χρήστη Microsoft Entra είτε ένα διακριτικό SPN Microsoft Entra

Έλεγχος ταυτότητας μιας περιόδου λειτουργίας δέσμης Livy API Spark με χρήση διακριτικού SPN Microsoft Entra

1. Δημιουργήστε ένα σημειωματάριο .ipynb στο Visual Studio Code και εισαγάγετε τον ακόλουθο κώδικα.

   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. Εκτελέστε το κελί σημειωματάριου, θα πρέπει να δείτε το διακριτικό Microsoft Entra να επιστρέφεται.

   

Έλεγχος ταυτότητας μιας περιόδου λειτουργίας Livy API Spark χρησιμοποιώντας ένα διακριτικό χρήστη Microsoft Entra

1. Δημιουργήστε ένα σημειωματάριο .ipynb στο Visual Studio Code και εισαγάγετε τον ακόλουθο κώδικα.

   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. Εκτελέστε το κελί σημειωματάριου και θα πρέπει να εμφανιστεί ένα αναδυόμενο παράθυρο στο πρόγραμμα περιήγησής σας, το οποίο σας επιτρέπει να επιλέξετε την ταυτότητα με την οποία θα εισέλθετε.

   

3. Αφού επιλέξετε την ταυτότητα με την οποία θα εισέλθετε, πρέπει να εγκρίνετε τα δικαιώματα API καταχώρησης εφαρμογής Microsoft Entra.

   

4. Κλείστε το παράθυρο του προγράμματος περιήγησης μετά την ολοκλήρωση του ελέγχου ταυτότητας.

   

5. Στο Visual Studio Code, θα πρέπει να δείτε το διακριτικό Microsoft Entra να επιστρέφεται.

   

Κατανόηση των πεδίων Code.* για το Livy API

Όταν οι εργασίες Spark εκτελούνται μέσω του Livy API, τα Code.* πεδία ελέγχουν σε ποιες εξωτερικές υπηρεσίες μπορεί να έχει πρόσβαση το Spark Runtime για λογαριασμό του χρήστη που έχει υποβληθεί σε έλεγχο ταυτότητας. Απαιτούνται δύο. Τα υπόλοιπα είναι προαιρετικά ανάλογα με τον φόρτο εργασίας σας.

Απαιτούμενος κώδικας.* πεδία

Προαιρετικός κώδικας.* πεδία

Προσθέστε αυτές τις εμβέλειες μόνο εάν οι εργασίες Spark πρέπει να έχουν πρόσβαση στις αντίστοιχες υπηρεσίες Azure κατά τον χρόνο εκτέλεσης.

Υποβάλετε μια παρτίδα Livy και παρακολουθήστε την εργασία παρτίδας.

1. Προσθέστε ένα άλλο κελί σημειωματάριου και εισαγάγετε αυτόν τον κώδικα.

   # 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. Εκτελέστε το κελί σημειωματάριου, θα πρέπει να δείτε πολλές γραμμές εκτυπωμένες καθώς δημιουργείται και εκτελείται η εργασία δέσμης Livy.

   

3. Για να δείτε τις αλλαγές, επιστρέψτε στο Lakehouse σας.

Ενοποίηση με περιβάλλοντα Fabric

Από προεπιλογή, αυτή η περίοδος λειτουργίας API Livy εκτελείται σε σχέση με την προεπιλεγμένη ομάδα εκκίνησης για τον χώρο εργασίας. Εναλλακτικά, μπορείτε να χρησιμοποιήσετε Fabric Περιβάλλοντα Δημιουργήστε, διαμορφώστε και χρησιμοποιήστε ένα περιβάλλον στο Microsoft Fabric για να προσαρμόσετε το χώρο συγκέντρωσης Spark που χρησιμοποιεί η περίοδος λειτουργίας Livy API για αυτές τις εργασίες Spark. Για να χρησιμοποιήσετε το περιβάλλον Fabric, ενημερώστε το προηγούμενο κελί σημειωματάριου με αυτήν την αλλαγή μίας γραμμής.

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

Προβολή των εργασιών σας στο Κέντρο παρακολούθησης

Μπορείτε να αποκτήσετε πρόσβαση στο Κέντρο παρακολούθησης για να προβάλετε διάφορες δραστηριότητες Apache Spark, επιλέγοντας Παρακολούθηση στις συνδέσεις περιήγησης στην αριστερή πλευρά.

1. Όταν ολοκληρωθεί η μαζική εργασία, μπορείτε να προβάλετε την κατάσταση της περιόδου λειτουργίας μεταβαίνοντας στην Παρακολούθηση.

   

2. Επιλέξτε και ανοίξτε το πιο πρόσφατο όνομα δραστηριότητας.

   

3. Σε αυτήν την περίπτωση περιόδου λειτουργίας Livy API, μπορείτε να δείτε την προηγούμενη υποβολή δέσμης, λεπτομέρειες εκτέλεσης, εκδόσεις Spark και ρυθμίσεις παραμέτρων. Παρατηρήστε την κατάσταση διακοπής στην επάνω δεξιά γωνία.

   

Για να ανακεφαλαιώσετε ολόκληρη τη διαδικασία, χρειάζεστε έναν απομακρυσμένο πελάτη όπως Visual Studio Code, ένα διακριτικό εφαρμογής Microsoft Entra, διεύθυνση URL τελικού σημείου Livy API, έλεγχο ταυτότητας στο Lakehouse σας, ένα ωφέλιμο φορτίο Spark στο Lakehouse σας και, τέλος, μια μαζική περίοδο λειτουργίας Livy API.

Σχετικό περιεχόμενο

• Τεκμηρίωση API REST Apache Livy
• Υποβολή εργασιών περιόδου λειτουργίας με χρήση του API Livy
• Επισκόπηση παρακολούθησης Apache Spark
• Λεπτομέρεια εφαρμογής Apache Spark