Freigeben über

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

  • Artikel

Hinweis

Die Livy-API für Fabric Data Engineering befindet sich in der öffentlichen Vorschau. Die Livy-API unterstützt azure Service Principal (SPN) nicht, obwohl wir beabsichtigen, den SPN-Support in Kürze hinzuzufügen.

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

Spark-Batch-Jobs mithilfe der Livy-API für die Fabric-Datenverarbeitung einreichen.

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: Lakehouse-Livy-Endpunkt und Verbindungszeichenfolge des Sitzungsauftrags

  3. Kopieren Sie die Verbindungszeichenfolge des Batchauftrags (zweites rotes Feld in der Abbildung) 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 <YourLakeHouseDataTableName>.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 (Azure Blob File System) der Nutzdaten, um in Ihrem Livy-API-Batchauftrag in Visual Studio Code darauf zu verweisen, und den Lakehouse-Tabellennamen in der SELECT-SQL-Anweisung.

    Screenshot: Python-Nutzdatenzelle

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

    Screenshot: Nutzdaten 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: Nutzdaten-ABFS-Pfad in den Eigenschaften der Datei im Lakehouse

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

Erstellen Sie eine Livy-API-Spark-Batchsitzung

  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 sollten Sie das zurückgegebene Microsoft Entra-Token sehen.

    Screenshot: Microsoft Entra-Token, das nach dem Ausführen der Zelle und dem Anmelden 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 Notebook-Zelle aus. Es sollten zwei Zeilen ausgegeben werden, wenn 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 Notebook-Zelle aus. Es sollten mehrere Zeilen ausgegeben werden, wenn der Livy Batch-Auftrag erstellt und ausgeführt wird.

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

  3. Gehen Sie zurück zu Ihrem Lakehouse, um die Änderungen zu sehen.

Integration in Fabric-Umgebungen

Diese Livy-API-Sitzung wird standardmäßig für den Standardstartpool für den Arbeitsbereich ausgeführt. Alternativ können Sie Fabric-Umgebungen zum Erstellen, Konfigurieren und Verwenden einer Umgebung in Microsoft Fabric verwenden, um den Spark-Pool anzupassen, den die Livy-API-Sitzung für diese Spark-Aufträge verwendet. Um Ihre Fabric-Umgebung zu verwenden, aktualisieren Sie einfach die vorherige Notebook-Zelle mit dieser einzelnen Zeilenänderung.

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

Anzeigen Ihrer Aufträge im Überwachungshub

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-Sitzungsszenario können Sie Ihre vorherigen Batch-Einsendungen, Ausführungsdetails, Spark-Versionen und Konfigurationen einsehen. Beachten Sie den Status „Angehalten“ oben rechts.

    Screenshot: Details der letzten Livy-API-Aktivität im Überwachungshub

Um den gesamten Prozess zusammenzufassen, benötigen Sie einen Remote-Client wie Visual Studio Code, ein Microsoft Entra-App-Token, eine Livy-API-Endpunkt-URL, die Authentifizierung für den Zugriff auf Ihr Lakehouse, eine Spark-Payload in Ihrem Lakehouse und schließlich eine Batch-Livy-API-Sitzung.

Zugehöriger Inhalt