Como criar e atualizar uma definição de trabalho do Spark com formato V2 por meio da API REST do Microsoft Fabric

O SJD (Definição de Trabalho do Spark) é um tipo de item do Fabric que permite que os usuários definam e executem trabalhos do Apache Spark no Microsoft Fabric. A API de Definição de Trabalho do Spark v2 permite que os usuários criem e atualizem itens de Definição de Trabalho do Spark com um novo formato chamado SparkJobDefinitionV2. O principal benefício de usar o formato v2 é que ele permite que os usuários gerenciem o arquivo executável principal e outros arquivos de biblioteca com uma única chamada de API, em vez de usar a API de armazenamento para carregar arquivos separadamente, nenhum token de armazenamento é necessário para gerenciar arquivos.

Pré-requisitos

• Um token do Microsoft Entra é necessário para acessar a API REST do Fabric. A biblioteca MSAL (Biblioteca de Autenticação da Microsoft) é recomendada para obter o token. Para obter mais informações, consulte o suporte ao fluxo de autenticação na MSAL.

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.

Visão geral do formato de definição de trabalho do Spark v2

No conteúdo do gerenciamento de um item de Definição de Trabalho do Spark, o campo definition é usado para especificar a configuração detalhada do item de Definição de Trabalho do Spark. O definition campo contém dois subcampos: format e parts. O campo format especifica o formato do item Definição de Trabalho Spark, que deve ser SparkJobDefinitionV2 para o formato v2.

O parts campo é uma matriz que contém a configuração detalhada do item de Definição de Trabalho do Spark. Cada item na parts matriz representa uma parte da configuração detalhada. Cada parte contém três subcampos: path, payloade payloadType. O path campo especifica o caminho da parte, o payload campo especifica o conteúdo da parte codificada em base64 e o payloadType campo especifica o tipo da carga, que deve ser InlineBase64.

Criar um item de definição de trabalho do Spark com o arquivo de definição principal e outros arquivos lib

No exemplo a seguir, criaremos um item de Definição de Trabalho do Spark que:

1. O nome é SJDHelloWorld.
2. O arquivo de definição principal é main.py, que lê um arquivo CSV de seu Lakehouse padrão para salvar como uma tabela Delta no mesmo Lakehouse.
3. Outro arquivo lib é libs.py, que tem uma função de utilitário para retornar o nome do arquivo CSV e da tabela Delta.
4. O Lakehouse padrão é definido como um ID de artefato Lakehouse específico.

O seguinte é o payload detalhado para criar o item de Definição de Trabalho do 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"
      }
    ]
  }
}

Para decodificar ou codificar a configuração detalhada, você pode usar as seguintes funções auxiliares no Python. Há também outras ferramentas online, como https://www.base64decode.org/, que podem realizar o mesmo trabalho.

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

Uma resposta de código HTTP 202 indica que o item Definição de Trabalho do Spark foi criado com êxito.

Obtenha a definição de tarefa do Spark com as partes da definição no formato v2

Com o novo formato v2, ao obter um item de Definição de Job do Spark com partes de definição, o conteúdo do arquivo de definição principal e outros arquivos lib são todos incluídos na resposta, codificados em base64 no campo parts. Aqui está um exemplo de como obter um item de Definição de Tarefa do Spark com componentes de definição:

1. Primeiro, faça uma solicitação POST para o ponto de extremidade https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{sjdartifactid}/getDefinitionParts?format=SparkJobDefinitionV2. Verifique se o valor do parâmetro de consulta de formato é SparkJobDefinitionV2.
2. Em seguida, nos cabeçalhos de resposta, verifique o código de status HTTP. Um código HTTP 202 indica que a solicitação foi aceita com êxito. Copie o valor x-ms-operation-id dos cabeçalhos de resposta.
3. Por fim, faça uma solicitação GET para o endpoint https://api.fabric.microsoft.com/v1/operations/{operationId} com o valor copiado x-ms-operation-id para obter o resultado da operação. No conteúdo da resposta, o campo definition contém a configuração detalhada do item da Definição do Trabalho Spark, incluindo o arquivo de definição principal e outros arquivos lib no campo parts.

Atualizar o item definição de trabalho do Spark com o arquivo de definição principal e outros arquivos lib no formato v2

Para atualizar um item de definição de trabalho do Spark existente com o arquivo de definição principal e outros arquivos lib no formato v2, você pode usar uma estrutura de conteúdo semelhante à operação de criação. Aqui está um exemplo de atualização do item de Definição de Trabalho do Spark criado na seção anterior:

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

Com a payload acima, as seguintes alterações são feitas nos arquivos:

1. O arquivo main.py é atualizado com novo conteúdo.
2. O lib1.py é excluído deste item de Definição de Trabalho do Spark e também removido do armazenamento do OneLake.
3. Um novo arquivo de lib2.py é adicionado a este item de Definição de Trabalho do Spark e carregado no armazenamento do OneLake.

Para atualizar o item de definição de trabalho do Spark, faça uma solicitação POST para o endpoint https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{sjdartifactid} com o payload acima. Uma resposta de código HTTP 202 indica que o item definição de trabalho do Spark foi atualizado com êxito.

Conteúdo relacionado

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