Come creare e aggiornare una definizione di lavoro Spark con il formato V2 tramite l'API Microsoft Fabric REST

La definizione del lavoro Spark (SJD) è un tipo di elemento Fabric che permette agli utenti di definire ed eseguire job Apache Spark in Fabric. L'API di definizione di job di Spark v2 consente agli utenti di creare e aggiornare gli elementi di definizione di job di Spark con un nuovo formato chiamato SparkJobDefinitionV2. Il vantaggio principale dell'uso del formato v2 consiste nel fatto che consente agli utenti di gestire il file eseguibile principale e altri file di libreria con una singola chiamata API, invece di usare l'API di archiviazione per caricare i file separatamente, non è necessario più token di archiviazione per la gestione dei file.

Prerequisiti

• Per accedere all'API REST di Fabric, è necessario un token Microsoft Entra. Per ottenere il token, è consigliabile usare la libreria MSAL (Libreria di Autenticazione Microsoft). Per altre informazioni, vedere Supporto del flusso di autenticazione in MSAL.

L'API Fabric REST definisce un endpoint unificato per le operazioni CRUD degli elementi Fabric. L'endpoint è https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items.

Panoramica del formato Definizione del lavoro Spark v2

Nel payload della gestione di un elemento di definizione del lavoro Spark, il definition campo viene utilizzato per specificare la configurazione dettagliata dell'elemento di definizione del lavoro Spark. Il definition campo contiene due sottocampi: format e parts. Il format campo specifica il formato dell'elemento di definizione del lavoro Spark, che dovrebbe essere SparkJobDefinitionV2 per il formato v2.

Il parts campo è un array che contiene la configurazione dettagliata dell'elemento di definizione del lavoro Spark. Ogni elemento nella parts matrice rappresenta una parte della configurazione dettagliata. Ogni parte contiene tre sottocampi: path, payloade payloadType. Il path campo specifica il percorso della parte, il payload campo specifica il contenuto della parte con codifica Base64 e il payloadType campo specifica il tipo del payload, che deve essere InlineBase64.

Crea un elemento di definizione del lavoro Spark con il file principale di definizione e altri file di lib

Nel seguente esempio, creeremo un elemento di definizione del lavoro Spark che:

1. Il nome è SJDHelloWorld.
2. Il file principale di definizione è main.py, che consiste nel leggere un file CSV dalla sua casa sul lago predefinita e salvarlo come tabella Delta nella stessa casa del lago.
3. L'altro file lib è libs.py, che ha una funzione di utilità per restituire il nome del file CSV e la tabella Delta.
4. La casa sul lago predefinita è impostata su un ID di elemento specifico della casa sul lago.

Di seguito è riportato il payload dettagliato per creare l'elemento di definizione del lavoro 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"
      }
    ]
  }
}

Per decodificare o codificare la configurazione dettagliata, è possibile usare le funzioni helper seguenti in Python. Ci sono anche altri strumenti online, https://www.base64decode.org/ ad esempio che possono eseguire lo stesso lavoro.

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

Una risposta con codice HTTP 202 indica che l'elemento di definizione del lavoro Spark è stato creato con successo.

Ottieni la definizione del lavoro Spark con parti di definizione sotto il formato v2

Con il nuovo formato v2, quando si riceve un elemento di definizione di un lavoro Spark con parti di definizione, il contenuto del file principale e degli altri file lib sono tutti inclusi nel payload di risposta, codificato in base64 sotto il parts campo. Ecco un esempio di come ottenere un elemento di definizione di lavoro Spark con parti di definizione:

1. Prima di tutto, effettuare una richiesta POST all'endpoint https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{sjditemid}/getDefinitionParts?format=SparkJobDefinitionV2. Assicurati che il valore del parametro di query 'format' sia SparkJobDefinitionV2.
2. Quindi, nelle intestazioni della risposta, controlla il codice di stato HTTP. Un codice HTTP 202 indica che la richiesta è stata accettata correttamente. Copiare il valore x-ms-operation-id dalle intestazioni della risposta.
3. Infine, effettuare una richiesta GET all'endpoint https://api.fabric.microsoft.com/v1/operations/{operationId} con il valore copiato x-ms-operation-id per ottenere il risultato dell'operazione. Nel payload di risposta, il definition campo contiene la configurazione dettagliata dell'elemento di definizione del lavoro Spark, includendo il file principale di definizione e altri file di lib sotto il parts campo.

Aggiorna l'elemento di definizione del lavoro Spark con il file principale di definizione e altri file lib sotto il formato v2

Per aggiornare un elemento di definizione di un lavoro Spark esistente con il file principale di definizione e altri file lib nel formato v2, puoi usare una struttura di payload simile a quella dell'operazione di creazione. Ecco un esempio di aggiornamento dell'elemento di definizione del lavoro di Spark creato nella sezione precedente:

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

Con il payload precedente, vengono apportate le modifiche seguenti ai file:

1. Il file main.py viene aggiornato con il nuovo contenuto.
2. Il lib1.py viene eliminato da questa voce di definizione del lavoro di Spark e anche rimosso dallo storage OneLake.
3. Un nuovo file lib2.py viene aggiunto a questo elemento di definizione del lavoro di Spark e caricato sullo storage OneLake.

Per aggiornare l'elemento di definizione del lavoro Spark, effettua una richiesta POST all'endpoint https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{sjditemid} con il payload sopra indicato. Una risposta HTTP code 202 indica che l'elemento di definizione del lavoro Spark è stato aggiornato con successo.

Contenuti correlati

• Pianificare ed eseguire una definizione di processo Apache Spark