Freigeben über

Verwenden der Livy-API zum Übermitteln und Ausführen von Livy-Batchaufträgen

  • Artikel

Hinweis

Die Livy-API für Fabric Datentechnik befindet sich in der Vorschau.

Gilt für:✅ Datentechnik und Data Science in Microsoft Fabric

Übermitteln von Spark-Batchaufträgen mit der Livy-API für Fabric Datentechnik.

Voraussetzungen

Die Livy-API definiert einen einheitlichen Endpunkt für Vorgänge. Ersetzen Sie die Platzhalter {Entra_TenantID}, {Entra_ClientID}, {Fabric_WorkspaceID} und {Fabric_LakehouseID} durch die entsprechenden Werte, wenn Sie den Beispielen in diesem Artikel folgen.

Konfigurieren von Visual Studio Code für Ihren Livy-API-Batch

  1. Wählen Sie Lakehouse-Einstellungen in Ihrem Fabric Lakehouse aus.

    Screenshot: Lakehouse-Einstellungen.

  2. Navigieren Sie zum Abschnitt Livy-Endpunkt.

    Screenshot des Lakehouse Livy-Endpunkts und der Verbindungszeichenfolge des Sitzungsauftrags.

  3. Kopieren Sie die Verbindungszeichenfolge des Batchauftrags (zweites rotes Feld im Bild) in Ihren Code.

  4. Navigieren Sie zum Microsoft Entra Admin Center, und kopieren Sie sowohl die Anwendungs-ID (Client-ID) als auch die Verzeichnis-ID (Mandant) in Ihren Code.

    Screenshot der Übersicht über die Livy-API-App im Microsoft Entra Admin Center.

Erstellen einer Spark-Nutzlast und Hochladen in Ihr Lakehouse

  1. Erstellen Sie ein .ipynb-Notebook in Visual Studio Code, und fügen Sie den folgenden Code ein

    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("livybatchdemo") 
       .getOrCreate())

     spark_context = spark_session.sparkContext
     spark_context.setLogLevel("DEBUG")  

     targetLakehouse = spark_context.getConf().get("spark.targetLakehouse")

     if targetLakehouse is not None:
       print("targetLakehouse: " + str(targetLakehouse))
     else:
       print("targetLakehouse is None")

df_valid_totalPrice = spark_session.sql("SELECT * FROM Guyhay_LivyDemo2.transactions where TotalPrice > 0")
df_valid_totalPrice_plus_year = df_valid_totalPrice.withColumn("transaction_year", col("TransactionDate").substr(1, 4))

deltaTablePath = "abfss:<YourABFSSpath>"+str(targetLakehouse)+".Lakehouse/Tables/CleanedTransactions"
df_valid_totalPrice_plus_year.write.mode('overwrite').format('delta').save(deltaTablePath)

  2. Speichern Sie die Python-Datei lokal. Diese Python-Codenutzlast enthält zwei Spark-Anweisungen, die an Daten in einem Lakehouse arbeiten und in Ihr Lakehouse hochgeladen werden müssen. Sie benötigen den ABFS-Pfad der Nutzlast, um in Ihrem Livy-API-Batchauftrag in Visual Studio Code zu referenzieren.

    Screenshot der Python-Nutzlastzelle.

  3. Laden Sie die Python-Nutzlast in den Dateiabschnitt Ihres Lakehouse hoch. > Daten abrufen > Dateien hochladen > klicken Sie das Feld „Dateien/Eingabe“.

    Screenshot der Nutzlast im Abschnitt „Dateien“ des Lakehouse.

  4. Nachdem sich die Datei im Abschnitt „Dateien“ Ihres Lakehouse befindet, klicken Sie rechts neben dem Dateinamen der Nutzlast auf die drei Punkte, und wählen Sie „Eigenschaften“ aus.

    Screenshot mit dem Nutzlast-ABFS-Pfad in den Eigenschaften der Datei im Lakehouse.

  5. Kopieren Sie diesen ABFS-Pfad in Ihre Notebookzelle in Schritt 1.

Erstellen einer Spark-Batchsitzung der Livy-API

  1. Erstellen Sie ein .ipynb-Notebook in Visual Studio Code, und fügen Sie den folgenden Code ein.

    
from msal import PublicClientApplication
import requests
import time

tenant_id = "<Entra_TenantID>"
client_id = "<Entra_ClientID>"

workspace_id = "<Fabric_WorkspaceID>"
lakehouse_id = "<Fabric_LakehouseID>"

app = PublicClientApplication(
   client_id,
   authority="https://login.microsoftonline.com/43a26159-4e8e-442a-9f9c-cb7a13481d48"
)

result = None

 # If no cached tokens or user interaction needed, acquire tokens interactively
 if not result:
     result = app.acquire_token_interactive(scopes=["https://api.fabric.microsoft.com/Lakehouse.Execute.All", "https://api.fabric.microsoft.com/Lakehouse.Read.All", "https://api.fabric.microsoft.com/Item.ReadWrite.All", 
                                                "https://api.fabric.microsoft.com/Workspace.ReadWrite.All", "https://api.fabric.microsoft.com/Code.AccessStorage.All", "https://api.fabric.microsoft.com/Code.AccessAzureKeyvault.All", 
                                                "https://api.fabric.microsoft.com/Code.AccessAzureDataExplorer.All", "https://api.fabric.microsoft.com/Code.AccessAzureDataLake.All", "https://api.fabric.microsoft.com/Code.AccessFabric.All"])

# Print the access token (you can use it to call APIs)
if "access_token" in result:
   print(f"Access token: {result['access_token']}")
else:
   print("Authentication failed or no access token obtained.")

if "access_token" in result:
   access_token = result['access_token']
   api_base_url_mist='https://api.fabric.microsoft.com/v1'
   livy_base_url = api_base_url_mist + "/workspaces/"+workspace_id+"/lakehouses/"+lakehouse_id +"/livyApi/versions/2023-12-01/batches"
   headers = {"Authorization": "Bearer " + access_token}

  2. Führen Sie das Notebookfenster aus. In Ihrem Browser sollte ein Popupfenster erscheinen, in dem Sie die Identität auswählen können, mit der Sie sich anmelden möchten.

    Screenshot des Anmeldebildschirms für die Microsoft Entra-App.

  3. Nachdem Sie die Identität ausgewählt haben, mit der Sie sich anmelden möchten, werden Sie auch aufgefordert, die API-Berechtigungen für die Microsoft Entra-App-Registrierung zu genehmigen.

    Screenshot der Microsoft Entra-App-API-Berechtigungen.

  4. Schließen Sie das Browserfenster nach Abschluss der Authentifizierung.

    Screenshot: Authentifizierung abgeschlossen.

  5. In Visual Studio Code sollte das Microsoft Entra-Token zurückgegeben werden.

    Screenshot des Microsoft Entra-Tokens, das nach dem Ausführen von Zelle und Anmeldung zurückgegeben wurde.

  6. Fügen Sie eine weitere Notebookzelle hinzu, und fügen Sie diesen Code ein.

    # call get batch API

get_livy_get_batch = livy_base_url
get_batch_response = requests.get(get_livy_get_batch, headers=headers)
if get_batch_response.status_code == 200:
   print("API call successful")
   print(get_batch_response.json())
else:
   print(f"API call failed with status code: {get_batch_response.status_code}")
   print(get_batch_response.text)

  7. Führen Sie die Notebookzelle aus. Es sollten zwei Zeilen gedruckt werden, während der Livy-Batchauftrag erstellt wird.

    Screenshot mit den Ergebnissen der Batchsitzungserstellung.

Übermitteln einer spark.sql-Anweisung mithilfe der Livy-API-Batchsitzung

  1. Fügen Sie eine weitere Notebookzelle hinzu, und fügen Sie diesen Code ein.

    # submit payload to existing batch session

print('Submit a spark job via the livy batch API to ') 

newlakehouseName = "YourNewLakehouseName"
create_lakehouse = api_base_url_mist + "/workspaces/" + workspace_id + "/items"
create_lakehouse_payload = {
   "displayName": newlakehouseName,
   "type": 'Lakehouse'
}

create_lakehouse_response = requests.post(create_lakehouse, headers=headers, json=create_lakehouse_payload)
print(create_lakehouse_response.json())

payload_data = {
   "name":"livybatchdemo_with"+ newlakehouseName,
   "file":"abfss://YourABFSPathToYourPayload.py", 
   "conf": {
     "spark.targetLakehouse": "Fabric_LakehouseID"
   }
}
get_batch_response = requests.post(get_livy_get_batch, headers=headers, json=payload_data)

print("The Livy batch job submitted successful")
print(get_batch_response.json())

  2. Führen Sie die Notebookzelle aus, es sollten mehrere Zeilen gedruckt werden, während der Livy Batch-Auftrag erstellt und ausgeführt wird.

    Screenshot mit Ergebnissen in Visual Studio Code, nachdem der Livy-Batchauftrag erfolgreich übermittelt wurde.

  3. Navigieren Sie zurück zu Ihrem Lakehouse, um die Änderungen anzuzeigen.

Anzeigen Ihrer Aufträge im Monitoring Hub

Sie können auf den Überwachungshub zugreifen, um verschiedene Apache Spark-Aktivitäten anzuzeigen, indem Sie in den Navigationslinks auf der linken Seite „Überwachen“ auswählen.

  1. Wenn der Batchauftrag abgeschlossen ist, können Sie den Sitzungsstatus anzeigen, indem Sie zu Monitor navigieren.

    Screenshot früherer Livy-API-Übermittlungen im Überwachungshub.

  2. Wählen Sie den Namen der letzten Aktivität aus, und öffnen Sie sie.

    Screenshot der letzten Livy-API-Aktivität im Monitoring Hub.

  3. In diesem Livy-API-Sitzungsfall können Sie Ihre vorherigen Batchübermittlung, Ausführungsdetails, Spark-Versionen und Konfigurationen einsehen. Beachten Sie den Status „Angehalten“ oben rechts.

    Screenshot der Details der letzten Livy-API-Aktivität im Monitoring Hub.

Um den gesamten Prozess erneut zu kapseln, benötigen Sie einen Remoteclient wie Visual Studio Code, ein Microsoft Entra-App-Token, eine Livy-API-Endpunkt-URL, die Authentifizierung über Ihr Lakehouse, eine Spark Payload in Ihrem Lakehouse und schließlich eine Batch-Livy-API-Sitzung.

Zugehöriger Inhalt