Como criar e atualizar uma definição de trabalho Spark com formato V2 através da API Microsoft Fabric REST

A definição de trabalho Spark (SJD) é um tipo de item Fabric que permite aos utilizadores definir e executar trabalhos Apache Spark no Fabric. A API de definição de trabalhos do Spark v2 permite aos utilizadores criar e atualizar itens de definição de trabalhos do Spark com um novo formato chamado SparkJobDefinitionV2. O principal benefício de usar o formato v2 é que permite aos utilizadores gerir o ficheiro executável principal e outros ficheiros da biblioteca com uma única chamada API; em vez de usar a API de armazenamento para carregar ficheiros separadamente, não é necessário mais token de armazenamento para gerir ficheiros.

Pré-requisitos

• É necessário um token Microsoft Entra para aceder à API REST do Fabric. Recomenda-se a biblioteca MSAL (Biblioteca de Autenticação da Microsoft) para obter o token. Para mais informações, veja Suporte a fluxo de autenticação em MSAL.

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

Visão geral do formato Spark Job Definition v2

Na carga útil de gerir um item de definição de trabalho Spark, o definition campo é usado para especificar a configuração detalhada do item de definição de trabalho Spark. O definition campo contém dois subcampos: format e parts. O format campo especifica o formato do item de definição do trabalho Spark, que deve ser SparkJobDefinitionV2 para o formato v2.

O parts campo é um array que contém a configuração detalhada do item de definição do trabalho do Spark. Cada elemento do parts array representa uma parte da configuração detalhada. Cada parte contém três subcampos: path, payload, e payloadType. O path campo especifica o caminho da peça, o payload campo especifica o conteúdo da peça codificada em base64, e o payloadType campo especifica o tipo da carga útil, que deve ser InlineBase64.

Crie um item de definição de trabalho no Spark com o ficheiro principal de definição e outros ficheiros de lib

No exemplo seguinte, vamos criar um item de definição de trabalho no Spark que:

1. O nome é SJDHelloWorld.
2. O ficheiro principal de definição é main.py, que consiste em ler um ficheiro CSV a partir da sua casa padrão e guardar como tabela Delta de volta na mesma casa do lago.
3. Outro ficheiro lib é libs.py, que tem uma função utilitária para devolver o nome do ficheiro CSV e da tabela Delta.
4. A casa do lago por defeito está definida para um ID de item específico da casa do lago.

Segue-se a carga útil detalhada para criar o item de definição do trabalho 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, pode usar as seguintes funções auxiliares em Python. Existem também outras ferramentas online, como https://www.base64decode.org/, que podem desempenhar a mesma função.

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 ao código HTTP 202 indica que o item de definição do trabalho Spark foi criado com sucesso.

Obtenha definição de trabalho Spark com partes de definição no formato v2

Com o novo formato v2, ao obter um item de definição de trabalho Spark com partes de definição, o conteúdo do ficheiro principal de definição e outros ficheiros lib são todos incluídos no payload de resposta, base64 codificado sob o parts campo. Aqui está um exemplo de obter um item de definição de trabalho Spark com partes de definição:

1. Primeiro, faça um pedido POST ao endpoint https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{sjditemid}/getDefinitionParts?format=SparkJobDefinitionV2. Certifique-se de que o valor do parâmetro de consulta de formato é SparkJobDefinitionV2.
2. Depois, nos cabeçalhos de resposta, verifica o código de estado HTTP. Um código HTTP 202 indica que o pedido foi aceite com sucesso. Copie o valor x-ms-operation-id dos cabeçalhos de resposta.
3. Finalmente, faça um pedido GET ao endpoint https://api.fabric.microsoft.com/v1/operations/{operationId} com o valor copiado x-ms-operation-id para obter o resultado da operação. Na carga útil de resposta, o definition campo contém a configuração detalhada do item de definição do trabalho Spark, incluindo o ficheiro principal de definição e outros ficheiros de lib sob o parts campo.

Atualize o item de definição do trabalho Spark com o ficheiro principal de definição e outros ficheiros de lib sob o formato v2

Para atualizar um item de definição de trabalho Spark existente com o ficheiro principal de definição e outros ficheiros lib sob o formato v2, pode usar uma estrutura de payload semelhante à da operação de criação. Aqui está um exemplo de atualização do item de definição de trabalho do Spark criado na secçã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 carga útil acima, são feitas as seguintes alterações aos ficheiros:

1. O ficheiro main.py é atualizado com novo conteúdo.
2. O lib1.py é eliminado deste item de definição de trabalho do Spark e também removido do armazenamento do OneLake.
3. Um novo ficheiro lib2.py é adicionado a este item de definição de trabalho do Spark e carregado para o armazenamento do OneLake.

Para atualizar o item de definição de trabalho do Spark, faça um pedido POST ao endpoint https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{sjditemid} com a carga útil acima. Uma resposta ao código HTTP 202 indica que o item de definição do trabalho Spark foi atualizado com sucesso.

Conteúdo relacionado

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