Como criar e atualizar uma definição de trabalho do Spark com a API Rest do Microsoft Fabric

Artigo
11/15/2023

A API REST do Microsoft Fabric fornece um ponto de extremidade de serviço para operações CRUD de itens do Fabric. Neste tutorial, examinamos um cenário de ponta a ponta de como criar e atualizar um artefato de definição de trabalho do Spark. Três etapas de alto nível estão envolvidas:

criar um item de definição de trabalho do Spark com algum estado inicial
carregar o arquivo de definição principal e outros arquivos de biblioteca
atualizar o item de definição de trabalho do Spark com a URL do OneLake do arquivo de definição principal e outros arquivos de biblioteca

Pré-requisitos

Um token do Microsoft Entra é necessário para acessar a API REST do Fabric. Recomenda-se a biblioteca MSAL para obter o token. Para obter mais informações, consulte Suporte ao fluxo de autenticação na MSAL.
Um token de armazenamento é necessário para acessar a API do OneLake. Para obter mais informações, consulte MSAL para Python.

Criar um item de definição de trabalho do Spark com o estado inicial

A API REST do Microsoft Fabric define um ponto de extremidade unificado para operações CRUD de itens do Fabric. O ponto de extremidade é https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items.

O detalhe do item é especificado dentro do corpo da solicitação. Aqui está um exemplo do corpo da solicitação para criar um item de definição de trabalho do Spark:

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

Neste exemplo, o item de definição de trabalho do Spark é nomeado como SJDHelloWorld. O campo payload é o conteúdo codificado em base64 da configuração de detalhes; após a decodificação, o conteúdo é:

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

Veja duas funções auxiliares para codificar e decodificar a configuração detalhada:

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

Este é o snippet de código para criar um item de definição de trabalho do 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)

Carregar o arquivo de definição principal e outros arquivos de biblioteca

Um token de armazenamento é necessário para carregar o arquivo no OneLake. Esta é uma função auxiliar para obter o token de armazenamento:


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

Agora que criamos um item de definição de trabalho do Spark, para torná-lo executável, precisamos configurar o arquivo de definição principal e as propriedades necessárias. O ponto de extremidade para carregar o arquivo para este item do SJD é https://onelake.dfs.fabric.microsoft.com/{workspaceId}/{sjdartifactid}. A mesma "workspaceId" da etapa anterior deve ser usada, o valor de "sjdartifactid" pode ser encontrado no corpo da resposta da etapa anterior. Este é o snippet de código para configurar o arquivo de definição principal:

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

Siga o mesmo processo para carregar os outros arquivos de biblioteca, se necessário.

Atualizar o item de definição de trabalho do Spark com a URL do OneLake do arquivo de definição principal e outros arquivos de biblioteca

Até agora, criamos um item de definição de trabalho do Spark com algum estado inicial, carregamos o arquivo de definição principal e outros arquivos de biblioteca. A última etapa é atualizar o item de definição de trabalho do Spark para definir as propriedades de URL do arquivo de definição principal e outros arquivos de biblioteca. O ponto de extremidade para atualizar o item de definição de trabalho do Spark é https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{sjdartifactid}. Os mesmos “workspaceId” e “sjdartifactid” das etapas anteriores devem ser usados. Este é o snippet de código para atualizar o item de definição de trabalho do 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)

Para recapitular todo o processo, a API REST do Fabric e a API do OneLake são necessárias para criar e atualizar um item de definição de trabalho do Spark. A API REST do Fabric é usada para criar e atualizar o item de definição de trabalho do Spark e a API do OneLake é usada para carregar o arquivo de definição principal e outros arquivos de biblioteca. O arquivo de definição principal e outros arquivos de biblioteca são carregados primeiro no OneLake. Em seguida, as propriedades da URL do arquivo de definição principal e outros arquivos de biblioteca são definidos no item de definição de trabalho do Spark.

Agendar e executar uma definição de trabalho do Apache Spark

Compartilhar via

Como criar e atualizar uma definição de trabalho do Spark com a API Rest do Microsoft Fabric

Pré-requisitos

Criar um item de definição de trabalho do Spark com o estado inicial

Carregar o arquivo de definição principal e outros arquivos de biblioteca

Atualizar o item de definição de trabalho do Spark com a URL do OneLake do arquivo de definição principal e outros arquivos de biblioteca

Comentários

Recursos adicionais

Compartilhar via

Como criar e atualizar uma definição de trabalho do Spark com a API Rest do Microsoft Fabric

Pré-requisitos

Criar um item de definição de trabalho do Spark com o estado inicial

Carregar o arquivo de definição principal e outros arquivos de biblioteca

Atualizar o item de definição de trabalho do Spark com a URL do OneLake do arquivo de definição principal e outros arquivos de biblioteca

Conteúdo relacionado

Comentários

Recursos adicionais