Delen via

Een Spark-taakdefinitie maken en bijwerken met microsoft Fabric Rest API

  • Artikel

De Microsoft Fabric Rest API biedt een service-eindpunt voor CRUD-bewerkingen van Fabric-items. In deze zelfstudie doorlopen we een end-to-end scenario voor het maken en bijwerken van een Spark-taakdefinitieartefact. Er zijn drie stappen op hoog niveau betrokken:

  1. een Spark-taakdefinitie-item met een bepaalde initiële status maken
  2. het hoofddefinitiebestand en andere lib-bestanden uploaden
  3. het Spark-taakdefinitie-item bijwerken met de OneLake-URL van het hoofddefinitiebestand en andere lib-bestanden

Vereisten

  1. Er is een Microsoft Entra-token vereist voor toegang tot de Fabric Rest API. De MSAL-bibliotheek wordt aanbevolen om het token op te halen. Zie ondersteuning voor verificatiestromen in MSAL voor meer informatie.
  2. Er is een opslagtoken vereist voor toegang tot de OneLake-API. Zie MSAL voor Python voor meer informatie.

Een Spark-taakdefinitie-item maken met de initiële status

De Rest API van Microsoft Fabric definieert een geïntegreerd eindpunt voor CRUD-bewerkingen van Fabric-items. Het eindpunt is https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items.

De itemdetails worden opgegeven in de hoofdtekst van de aanvraag. Hier volgt een voorbeeld van de aanvraagbody voor het maken van een Spark-taakdefinitie-item:

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

In dit voorbeeld krijgt het Spark-taakdefinitie-item de naam SJDHelloWorld. Het payload veld is de base64-gecodeerde inhoud van de detailconfiguratie, na het decoderen, de inhoud is:

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

Hier volgen twee helperfuncties voor het coderen en decoderen van de gedetailleerde installatie:

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

Dit is het codefragment voor het maken van een Spark-taakdefinitie-item:

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)

Het hoofddefinitiebestand en andere lib-bestanden uploaden

Er is een opslagtoken vereist om het bestand te uploaden naar OneLake. Hier volgt een helperfunctie om het opslagtoken op te halen:


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']

We hebben nu een Spark-taakdefinitie-item gemaakt, zodat het kan worden uitgevoerd. We moeten het hoofddefinitiebestand en de vereiste eigenschappen instellen. Het eindpunt voor het uploaden van het bestand voor dit SJD-item is https://onelake.dfs.fabric.microsoft.com/{workspaceId}/{sjdartifactid}. Dezelfde 'workspaceId' uit de vorige stap moet worden gebruikt. De waarde van 'sjdartifactid' kan worden gevonden in de antwoordtekst van de vorige stap. Hier volgt het codefragment voor het instellen van het hoofddefinitiebestand:

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())

Volg hetzelfde proces om de andere lib-bestanden indien nodig te uploaden.

Werk het Spark-taakdefinitie-item bij met de OneLake-URL van het hoofddefinitiebestand en andere lib-bestanden

Tot nu toe hebben we een Spark-taakdefinitie-item gemaakt met een bepaalde initiële status, het hoofddefinitiebestand en andere lib-bestanden geüpload. De laatste stap is het bijwerken van het Spark-taakdefinitie-item om de URL-eigenschappen van het hoofddefinitiebestand en andere lib-bestanden in te stellen. Het eindpunt voor het bijwerken van het Spark-taakdefinitie-item is https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{sjdartifactid}. Dezelfde 'workspaceId' en 'sjdartifactid' uit de vorige stappen moeten worden gebruikt. Dit is het codefragment voor het bijwerken van het Spark-taakdefinitie-item:


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)

Als u het hele proces wilt samenvatten, zijn zowel Fabric REST API als OneLake-API nodig om een Spark-taakdefinitie-item te maken en bij te werken. De Fabric REST API wordt gebruikt om het Spark-taakdefinitie-item te maken en bij te werken. De OneLake-API wordt gebruikt om het hoofddefinitiebestand en andere lib-bestanden te uploaden. Het hoofddefinitiebestand en andere lib-bestanden worden eerst geüpload naar OneLake. Vervolgens worden de URL-eigenschappen van het hoofddefinitiebestand en andere lib-bestanden ingesteld in het Spark-taakdefinitie-item.

Gerelateerde inhoud