Come creare e aggiornare una definizione di processo Spark con l'API REST di Microsoft Fabric

Articolo
12/06/2023

L'API REST di Microsoft Fabric fornisce un endpoint di servizio per le operazioni CRUD degli elementi di Fabric. In questa esercitazione viene illustrato uno scenario end-to-end di come creare e aggiornare un artefatto di definizione del processo Spark. Sono coinvolti tre passaggi generali:

creare un elemento Definizione processo Spark con uno stato iniziale
caricare il file di definizione principale e altri file lib
aggiornare l'elemento Definizione processo Spark con l'URL OneLake del file di definizione principale e altri file lib

Prerequisiti

Per accedere all'API REST di Fabric, è necessario un token Microsoft Entra. La libreria MSAL è consigliata per ottenere il token. Per altre informazioni, vedere Supporto del flusso di autenticazione in MSAL.
Per accedere all'API OneLake è necessario un token di archiviazione. Per altre informazioni, vedere MSAL per Python.

Creare un elemento Definizione processo Spark con lo stato iniziale

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

Il dettaglio dell'elemento viene specificato all'interno del corpo della richiesta. Di seguito è riportato un esempio del corpo della richiesta per la creazione di un elemento Di definizione processo Spark:

{
    "displayName": "SJDHelloWorld",
    "type": "SparkJobDefinition",
    "definition": {
        "format": "SparkJobDefinitionV1",
        "parts": [
            {
                "path": "SparkJobDefinitionV1.json",
                "payload":"eyJleGVjdXRhYmxlRmlsZSI6bnVsbCwiZGVmYXVsdExha2Vob3VzZUFydGlmYWN0SWQiOiIiLCJtYWluQ2xhc3MiOiIiLCJhZGRpdGlvbmFsTGFrZWhvdXNlSWRzIjpbXSwicmV0cnlQb2xpY3kiOm51bGwsImNvbW1hbmRMaW5lQXJndW1lbnRzIjoiIiwiYWRkaXRpb25hbExpYnJhcnlVcmlzIjpbXSwibGFuZ3VhZ2UiOiIiLCJlbnZpcm9ubWVudEFydGlmYWN0SWQiOm51bGx9",
                "payloadType": "InlineBase64"
            }
        ]
    }
}

In questo esempio l'elemento Definizione processo Spark viene denominato come SJDHelloWorld. Il payload campo è il contenuto codificato in base64 della configurazione dei dettagli, dopo la decodifica, il contenuto è:

{
    "executableFile":null,
    "defaultLakehouseArtifactId":"",
    "mainClass":"",
    "additionalLakehouseIds":[],
    "retryPolicy":null,
    "commandLineArguments":"",
    "additionalLibraryUris":[],
    "language":"",
    "environmentArtifactId":null
}

Ecco due funzioni helper per codificare e decodificare la configurazione dettagliata:

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

Ecco il frammento di codice per creare un elemento definizione processo Spark:

import requests

bearerToken = "breadcrumb"; # replace this token with the real AAD token

headers = {
    "Authorization": f"Bearer {bearerToken}", 
    "Content-Type": "application/json"  # Set the content type based on your request
}

payload = "eyJleGVjdXRhYmxlRmlsZSI6bnVsbCwiZGVmYXVsdExha2Vob3VzZUFydGlmYWN0SWQiOiIiLCJtYWluQ2xhc3MiOiIiLCJhZGRpdGlvbmFsTGFrZWhvdXNlSWRzIjpbXSwicmV0cnlQb2xpY3kiOm51bGwsImNvbW1hbmRMaW5lQXJndW1lbnRzIjoiIiwiYWRkaXRpb25hbExpYnJhcnlVcmlzIjpbXSwibGFuZ3VhZ2UiOiIiLCJlbnZpcm9ubWVudEFydGlmYWN0SWQiOm51bGx9"

# Define the payload data for the POST request
payload_data = {
    "displayName": "SJDHelloWorld",
    "Type": "SparkJobDefinition",
    "definition": {
        "format": "SparkJobDefinitionV1",
        "parts": [
            {
                "path": "SparkJobDefinitionV1.json",
                "payload": payload,
                "payloadType": "InlineBase64"
            }
        ]
    }
}

# Make the POST request with Bearer authentication
sjdCreateUrl = f"https://api.fabric.microsoft.com//v1/workspaces/{workspaceId}/items"
response = requests.post(sjdCreateUrl, json=payload_data, headers=headers)

Caricare il file di definizione principale e altri file lib

Per caricare il file in OneLake, è necessario un token di archiviazione. Ecco una funzione helper per ottenere il token di archiviazione:


import msal

def getOnelakeStorageToken():
    app = msal.PublicClientApplication(
        "{client id}", # this filed should be the client id 
        authority="https://login.microsoftonline.com/microsoft.com")

    result = app.acquire_token_interactive(scopes=["https://storage.azure.com/.default"])

    print(f"Successfully acquired AAD token with storage audience:{result['access_token']}")

    return result['access_token']

A questo punto è stato creato un elemento Definizione processo Spark, per renderlo eseguibile, è necessario configurare il file di definizione principale e le proprietà necessarie. L'endpoint per il caricamento del file per questo elemento SJD è https://onelake.dfs.fabric.microsoft.com/{workspaceId}/{sjdartifactid}. È consigliabile usare lo stesso "workspaceId" del passaggio precedente, il valore di "sjdartifactid" può essere trovato nel corpo della risposta del passaggio precedente. Ecco il frammento di codice per configurare il file di definizione principale:

import requests

# three steps are required: create file, append file, flush file

onelakeEndPoint = "https://onelake.dfs.fabric.microsoft.com/workspaceId/sjdartifactid"; # replace the id of workspace and artifact with the right one
mainExecutableFile = "main.py"; # the name of the main executable file
mainSubFolder = "Main"; # the sub folder name of the main executable file. Don't change this value


onelakeRequestMainFileCreateUrl = f"{onelakeEndPoint}/{mainSubFolder}/{mainExecutableFile}?resource=file" # the url for creating the main executable file via the 'file' resource type
onelakePutRequestHeaders = {
    "Authorization": f"Bearer {onelakeStorageToken}", # the storage token can be achieved from the helper function above
}

onelakeCreateMainFileResponse = requests.put(onelakeRequestMainFileCreateUrl, headers=onelakePutRequestHeaders)
if onelakeCreateMainFileResponse.status_code == 201:
    # Request was successful
    print(f"Main File '{mainExecutableFile}' was successfully created in onelake.")

# with previous step, the main executable file is created in OneLake, now we need to append the content of the main executable file

appendPosition = 0;
appendAction = "append";

### Main File Append.
mainExecutableFileSizeInBytes = 83; # the size of the main executable file in bytes
onelakeRequestMainFileAppendUrl = f"{onelakeEndPoint}/{mainSubFolder}/{mainExecutableFile}?position={appendPosition}&action={appendAction}";
mainFileContents = "filename = 'Files/' + Constant.filename; tablename = 'Tables/' + Constant.tablename"; # the content of the main executable file, please replace this with the real content of the main executable file
mainExecutableFileSizeInBytes = 83; # the size of the main executable file in bytes, this value should match the size of the mainFileContents

onelakePatchRequestHeaders = {
    "Authorization": f"Bearer {onelakeStorageToken}",
    "Content-Type" : "text/plain"
}

onelakeAppendMainFileResponse = requests.patch(onelakeRequestMainFileAppendUrl, data = mainFileContents, headers=onelakePatchRequestHeaders)
if onelakeAppendMainFileResponse.status_code == 202:
    # Request was successful
    print(f"Successfully Accepted Main File '{mainExecutableFile}' append data.")

# with previous step, the content of the main executable file is appended to the file in OneLake, now we need to flush the file

flushAction = "flush";

### Main File flush
onelakeRequestMainFileFlushUrl = f"{onelakeEndPoint}/{mainSubFolder}/{mainExecutableFile}?position={mainExecutableFileSizeInBytes}&action={flushAction}"
print(onelakeRequestMainFileFlushUrl)
onelakeFlushMainFileResponse = requests.patch(onelakeRequestMainFileFlushUrl, headers=onelakePatchRequestHeaders)
if onelakeFlushMainFileResponse.status_code == 200:
    print(f"Successfully Flushed Main File '{mainExecutableFile}' contents.")
else:
    print(onelakeFlushMainFileResponse.json())

Seguire lo stesso processo per caricare gli altri file lib, se necessario.

Aggiornare l'elemento Definizione processo Spark con l'URL OneLake del file di definizione principale e altri file lib

Fino ad ora, è stato creato un elemento Definizione processo Spark con uno stato iniziale, caricato il file di definizione principale e altri file lib, L'ultimo passaggio consiste nell'aggiornare l'elemento Definizione processo Spark per impostare le proprietà URL del file di definizione principale e altri file lib. L'endpoint per l'aggiornamento dell'elemento Definizione processo Spark è https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{sjdartifactid}. È necessario usare lo stesso "workspaceId" e "sjdartifactid" dei passaggi precedenti. Ecco il frammento di codice per aggiornare l'elemento Definizione processo Spark:


mainAbfssPath = f"abfss://{workspaceId}@onelake.dfs.fabric.microsoft.com/{sjdartifactid}/Main/{mainExecutableFile}" # the workspaceId and sjdartifactid are the same as previous steps, the mainExecutableFile is the name of the main executable file
libsAbfssPath = f"abfss://{workspaceId}@onelake.dfs.fabric.microsoft.com/{sjdartifactid}/Libs/{libsFile}"  # the workspaceId and sjdartifactid are the same as previous steps, the libsFile is the name of the libs file
defaultLakehouseId = 'defaultLakehouseid'; # replace this with the real default lakehouse id

updateRequestBodyJson = {
    "executableFile":mainAbfssPath,
    "defaultLakehouseArtifactId":defaultLakehouseId,
    "mainClass":"",
    "additionalLakehouseIds":[],
    "retryPolicy":None,
    "commandLineArguments":"",
    "additionalLibraryUris":[libsAbfssPath],
    "language":"Python",
    "environmentArtifactId":None}

# Encode the bytes as a Base64-encoded string
base64EncodedUpdateSJDPayload = json_to_base64(updateRequestBodyJson)

# Print the Base64-encoded string
print("Base64-encoded JSON payload for SJD Update:")
print(base64EncodedUpdateSJDPayload)

# Define the API URL
updateSjdUrl = f"https://api.fabric.microsoft.com//v1/workspaces/{workspaceId}/items/{sjdartifactid}/updateDefinition"

updatePayload = base64EncodedUpdateSJDPayload
payloadType = "InlineBase64"
path = "SparkJobDefinitionV1.json"
format = "SparkJobDefinitionV1"
Type = "SparkJobDefinition"

# Define the headers with Bearer authentication
bearerToken = "breadcrumb"; # replace this token with the real AAD token

headers = {
    "Authorization": f"Bearer {bearerToken}", 
    "Content-Type": "application/json"  # Set the content type based on your request
}

# Define the payload data for the POST request
payload_data = {
    "displayName": "sjdCreateTest11",
    "Type": Type,
    "definition": {
        "format": format,
        "parts": [
            {
                "path": path,
                "payload": updatePayload,
                "payloadType": payloadType
            }
        ]
    }
}


# Make the POST request with Bearer authentication
response = requests.post(updateSjdUrl, json=payload_data, headers=headers)
if response.status_code == 200:
    print("Successfully updated SJD.")
else:
    print(response.json())
    print(response.status_code)

Per riepilogare l'intero processo, sono necessari sia l'API REST fabric che l'API OneLake per creare e aggiornare un elemento definizione processo Spark. L'API REST fabric viene usata per creare e aggiornare l'elemento Definizione processo Spark, l'API OneLake viene usata per caricare il file di definizione principale e altri file lib. Il file di definizione principale e altri file lib vengono prima caricati in OneLake. Le proprietà URL del file di definizione principale e altri file lib vengono quindi impostate nell'elemento Definizione processo Spark.

Pianificare ed eseguire una definizione di processo Apache Spark

Condividi tramite

Come creare e aggiornare una definizione di processo Spark con l'API REST di Microsoft Fabric

Prerequisiti

Creare un elemento Definizione processo Spark con lo stato iniziale

Caricare il file di definizione principale e altri file lib

Aggiornare l'elemento Definizione processo Spark con l'URL OneLake del file di definizione principale e altri file lib

Commenti e suggerimenti

Risorse aggiuntive

Condividi tramite

Come creare e aggiornare una definizione di processo Spark con l'API REST di Microsoft Fabric

Prerequisiti

Creare un elemento Definizione processo Spark con lo stato iniziale

Caricare il file di definizione principale e altri file lib

Aggiornare l'elemento Definizione processo Spark con l'URL OneLake del file di definizione principale e altri file lib

Contenuto correlato

Commenti e suggerimenti

Risorse aggiuntive