Comment créer et mettre à jour une définition de job Spark avec le format V2 via l’API Microsoft Fabric REST

La définition des tâches Spark (SJD) est un type d’élément Fabric qui permet aux utilisateurs de définir et d’exécuter des tâches Apache Spark dans Fabric. L’API de définition de tâches Spark v2 permet aux utilisateurs de créer et de mettre à jour des éléments de définition de tâches Spark avec un nouveau format appelé SparkJobDefinitionV2. L’avantage principal de l’utilisation du format v2 est qu’il permet aux utilisateurs de gérer le fichier exécutable principal et d’autres fichiers de bibliothèque avec un seul appel d’API, au lieu d’utiliser l’API de stockage pour charger des fichiers séparément, aucun jeton de stockage n’est nécessaire pour gérer les fichiers.

Prerequisites

L’API Fabric REST définit un point de terminaison unifié pour les opérations CRUD des éléments Fabric. Le point de terminaison a la valeur https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items.

Aperçu du format Spark Job Definition v2

Dans la charge utile de gestion d’un élément de définition de la tâche Spark, le definition champ sert à spécifier la configuration détaillée de l’élément de définition de la tâche Spark. Le definition champ contient deux sous-champs : format et parts. Le format champ spécifie le format de l’élément de définition du job Spark, qui doit être SparkJobDefinitionV2 pour le format v2.

Le parts champ est un tableau qui contient la configuration détaillée de l’élément de définition du travail Spark. Chaque élément du parts tableau représente une partie de la configuration détaillée. Chaque partie contient trois sous-champs : path, payloadet payloadType. Le path champ spécifie le chemin d’accès de la partie, le payload champ spécifie le contenu de la partie encodé en base64, et le payloadType champ spécifie le type de la charge utile, qui doit être InlineBase64.

Important

Ce format v2 ne prend en charge que les définitions de tâches Spark avec des formats de fichiers .py ou .scala. Le format de fichier .jar n’est pas pris en charge.

Créez un élément de définition de job Spark avec le fichier principal de définition et d’autres fichiers de libération

Dans l’exemple suivant, nous allons créer un élément de définition de job Spark qui suit :

  1. Le nom est SJDHelloWorld.
  2. Le fichier principal de définition est main.py, qui consiste à lire un fichier CSV depuis sa maison de lac par défaut et à enregistrer comme table Delta dans la même maison de lac.
  3. Un autre fichier lib est libs.py, qui a une fonction utilitaire pour renvoyer le nom du fichier CSV et la table Delta.
  4. La maison par défaut est définie sur un identifiant d’élément spécifique de la maison du lac.

Voici la charge utile détaillée pour créer l’élément de définition du travail Spark.

{
  "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"
      }
    ]
  }
}

Pour décoder ou encoder la configuration détaillée, vous pouvez utiliser les fonctions d’assistance suivantes dans Python. Il existe également d’autres outils en ligne tels que https://www.base64decode.org/ ceux qui peuvent effectuer le même travail.

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

Une réponse HTTP code 202 indique que l’élément de définition du travail Spark a été créé avec succès.

Obtenez la définition du job Spark avec les parties de définition sous le format v2

Avec le nouveau format v2, lors de l’obtention d’un élément de définition du job Spark avec des parties de définition, le contenu du fichier principal et d’autres fichiers lib sont tous inclus dans la charge utile de réponse, encodée base64 sous le parts champ. Voici un exemple d’obtention d’un élément de définition de job Spark avec des parties de définition :

  1. Tout d’abord, effectuez une requête POST au point de terminaison https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{sjditemid}/getDefinitionParts?format=SparkJobDefinitionV2. Vérifiez que la valeur du paramètre de requête de format est SparkJobDefinitionV2.
  2. Ensuite, dans les en-têtes de réponse, vérifiez le code d’état HTTP. Un code HTTP 202 indique que la requête a été acceptée avec succès. Copiez la valeur x-ms-operation-id des en-têtes de réponse.
  3. Enfin, effectuez une requête GET au point de terminaison https://api.fabric.microsoft.com/v1/operations/{operationId} avec la valeur copiée x-ms-operation-id pour obtenir le résultat de l’opération. Dans la charge utile de réponse, le definition champ contient la configuration détaillée de l’élément de définition du travail Spark, incluant le fichier principal de définition et d’autres fichiers de lib sous le parts champ.

Mettre à jour l’élément de définition du job Spark avec le fichier principal de définition et d’autres fichiers de lib sous le format v2

Pour mettre à jour un élément de définition de travail Spark existant avec le fichier principal de définition et d’autres fichiers de lib sous le format v2, vous pouvez utiliser une structure de charge utile similaire à celle de l’opération create. Voici un exemple de mise à jour de l’élément de définition du job Spark créé dans la section précédente :

{
  "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"
      }
    ]
  }
}

Avec la charge utile ci-dessus, les modifications suivantes sont apportées aux fichiers :

  1. Le fichier main.py est mis à jour avec le nouveau contenu.
  2. Le lib1.py est supprimé de cet élément de définition du job Spark et également retiré du stockage OneLake.
  3. Un nouveau fichier lib2.py est ajouté à cet élément de définition de la tâche Spark et téléchargé dans le stockage OneLake.

Pour mettre à jour l’élément de définition de la tâche Spark, effectuez une requête POST au point de terminaison https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{sjditemid} avec la charge utile ci-dessus. Une réponse HTTP code 202 indique que l’élément de définition du travail Spark a été mis à jour avec succès.

Contenu connexe

  • Last updated on