Wie man eine Spark-Jobdefinition mit V2-Format über die Microsoft Fabric REST API erstellt und aktualisiert

Die Spark-Job-Definition (SJD) ist eine Art von Fabric-Item, die es Nutzern ermöglicht, Apache Spark-Jobs in Fabric zu definieren und auszuführen. Die Spark-Job-Definitions-API v2 ermöglicht es Benutzern, Spark-Job-Definitionselemente mit einem neuen Format namens SparkJobDefinitionV2zu erstellen und zu aktualisieren. Der Hauptvorteil der Verwendung des v2-Formats besteht darin, dass Benutzer die haupt ausführbare Datei und andere Bibliotheksdateien mit einem einzigen API-Aufruf verwalten können, anstatt die Speicher-API zum getrennten Hochladen von Dateien zu verwenden, für die Verwaltung von Dateien kein weiteres Speichertoken erforderlich ist.

Voraussetzungen

• Für den Zugriff auf die Fabric-REST-API ist ein Microsoft Entra-Token erforderlich. Die Verwendung der MSAL-Bibliothek (Microsoft Authentication Library (MSAL)) wird empfohlen, um das Token abzurufen. Weitere Informationen finden Sie unter Unterstützung des Authentifizierungsflusses in MSAL.

Die Fabric REST API definiert einen einheitlichen Endpunkt für CRUD-Operationen von Fabric-Artikeln. Der Endpunkt ist https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items.

Überblick über das Spark-Job-Definition v2 Format

Im Payload zur Verwaltung eines Spark-Job-Definitionselements wird das definition Feld verwendet, um die detaillierte Einrichtung des Spark-Job-Definitionselements anzugeben. Das definition Feld enthält zwei Unterfelder: format und parts. Das Feld format spezifiziert das Format des Spark-Job-Definitionselements, das für das v2-Format vorgesehen ist SparkJobDefinitionV2 .

Das Feld parts ist ein Array, das die detaillierte Konfiguration des Spark-Job-Definitionselements enthält. Jedes Element im parts Array stellt einen Teil des detaillierten Setups dar. Jeder Teil enthält drei Unterfelder: path, , payloadund payloadType. Das path-Feld spezifiziert den Pfad des Teils, das payload-Feld spezifiziert den Inhalt des Teils, der base64-codiert ist, und das payloadType-Feld gibt den Typ der Nutzlast an, welcher InlineBase64 sein sollte.

Erstellen Sie ein Spark-Job-Definitionselement mit der Hauptdefinitionsdatei und anderen Bibliotheksdateien

Im folgenden Beispiel erstellen wir ein Spark-Job-Definitionselement, das:

1. Name ist SJDHelloWorld.
2. Die Hauptdefinitionsdatei ist main.py, was bedeutet, eine CSV-Datei aus ihrem Standard-Lakehouse zu lesen und als Delta-Tabelle zurück im selben Lakehouse zu speichern.
3. Andere Lib-Datei ist libs.py, die über eine Hilfsfunktion verfügt, um den Namen der CSV-Datei und der Delta-Tabelle zurückzugeben.
4. Das Standard-Lakehouse ist auf eine bestimmte Lakehouse-Item-ID gesetzt.

Im Folgenden finden Sie die detaillierte Nutzlast zur Erstellung des Spark-Job-Definitionselements.

{
  "displayName": "SJDHelloWorld",
  "type": "SparkJobDefinition",
  "definition": {
    "format": "SparkJobDefinitionV2",
    "parts": [
      {
        "path": "SparkJobDefinitionV1.json",
        "payload": "<REDACTED>",
        "payloadType": "InlineBase64"
      },
      {
        "path": "Main/main.py",
        "payload": "<REDACTED>",
        "payloadType": "InlineBase64"
      },
      {
        "path": "Libs/lib1.py",
        "payload": "<REDACTED>",
        "payloadType": "InlineBase64"
      }
    ]
  }
}

Um das detaillierte Setup zu decodieren oder zu codieren, können Sie die folgenden Hilfsfunktionen in Python verwenden. Es gibt auch andere Onlinetools, z. B. https://www.base64decode.org/, die die gleiche Aufgabe ausführen können.

import base64

def json_to_base64(json_data):
    # Serialize the JSON data to a string
    json_string = json.dumps(json_data)
    
    # Encode the JSON string as bytes
    json_bytes = json_string.encode('utf-8')
    
    # Encode the bytes as Base64
    base64_encoded = base64.b64encode(json_bytes).decode('utf-8')
    
    return base64_encoded

def base64_to_json(base64_data):
    # Decode the Base64-encoded string to bytes
    base64_bytes = base64_data.encode('utf-8')
    
    # Decode the bytes to a JSON string
    json_string = base64.b64decode(base64_bytes).decode('utf-8')
    
    # Deserialize the JSON string to a Python dictionary
    json_data = json.loads(json_string)
    
    return json_data

Eine HTTP-Code-202-Antwort zeigt an, dass das Spark-Job-Definitionselement erfolgreich erstellt wurde.

Erhalten Sie die Spark-Job-Definition mit Definitionsteilen im v2-Format

Mit dem neuen v2-Format werden beim Erhalt eines Spark-Job-Definitionselements mit Definitionsteilen der Dateiinhalte der Hauptdefinitionsdatei und anderer Bibliotheksdateien alle in die Antwort-Payload einbezogen, base64 unter dem Feld parts codiert. Hier ist ein Beispiel, wie man ein Spark-Job-Definitionselement mit Definitionsteilen bekommt:

1. Stellen Sie zunächst eine POST-Anforderung an den Endpunkt https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{sjditemid}/getDefinitionParts?format=SparkJobDefinitionV2. Stellen Sie sicher, dass der Wert des Formatabfrageparameters lautet SparkJobDefinitionV2.
2. Überprüfen Sie dann in den Antwortheadern den HTTP-Statuscode. Ein HTTP-Code 202 gibt an, dass die Anforderung erfolgreich akzeptiert wurde. Kopieren Sie den x-ms-operation-id Wert aus den Antwortheadern.
3. Nehmen Sie schließlich eine GET-Anforderung an den Endpunkt https://api.fabric.microsoft.com/v1/operations/{operationId} mit dem kopierten x-ms-operation-id Wert vor, um das Vorgangsergebnis abzurufen. Im Antwortpayload enthält das Feld definition die detaillierte Einrichtung des Spark-Job-Definitionselements, einschließlich der Hauptdefinitionsdatei und anderer Bibliotheksdateien unter dem Feld parts .

Aktualisieren Sie das Spark-Job-Definitionselement mit der Hauptdefinitionsdatei und anderen Bibliotheksdateien im v2-Format

Um ein bestehendes Spark-Job-Definitionselement mit der Hauptdefinitionsdatei und anderen Bibliotheksdateien im v2-Format zu aktualisieren, kann man eine ähnliche Payload-Struktur wie die Erstellungsoperation verwenden. Hier ist ein Beispiel für die Aktualisierung des Spark-Job-Definitionselements, das im vorherigen Abschnitt erstellt wurde:

{
  "displayName": "SJDHelloWorld",
  "type": "SparkJobDefinition",
  "definition": {
    "format": "SparkJobDefinitionV2",
    "parts": [
      {
        "path": "SparkJobDefinitionV1.json",
        "payload": "<REDACTED>",
        "payloadType": "InlineBase64"
      },
      {
        "path": "Main/main.py",
        "payload": "<REDACTED>",
        "payloadType": "InlineBase64"
      },
      {
        "path": "Libs/lib2.py",
        "payload": "<REDACTED>",
        "payloadType": "InlineBase64"
      }
    ]
  }
}

Mit der obigen Nutzlast werden die folgenden Änderungen an den Dateien vorgenommen:

1. Die main.py Datei wird mit neuem Inhalt aktualisiert.
2. Der lib1.py wird aus diesem Spark-Job-Definitionselement gelöscht und auch aus dem OneLake-Speicher entfernt.
3. Eine neue lib2.py-Datei wird diesem Spark-Job-Definitionselement hinzugefügt und in den OneLake-Speicher hochgeladen.

Um das Spark-Job-Definitionselement zu aktualisieren, stellen Sie eine POST-Anfrage an den Endpunkt https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{sjditemid} mit der oben genannten Nutzlast. Eine HTTP-Code-202-Antwort zeigt an, dass das Spark-Job-Definitionselement erfolgreich aktualisiert wurde.

Verwandte Inhalte

• Planen und Ausführen einer Apache Spark-Auftragsdefinition