Cómo crear y actualizar una definición de trabajo en Spark con formato V2 mediante la API REST de Microsoft Fabric

La definición de trabajos Spark (SJD) es un tipo de elemento Fabric que permite a los usuarios definir y ejecutar trabajos Apache Spark en Fabric. La API de definición de trabajos de Spark v2 permite a los usuarios crear y actualizar elementos de definición de trabajos de Spark con un nuevo formato llamado SparkJobDefinitionV2. La principal ventaja de usar el formato v2 es que permite a los usuarios administrar el archivo ejecutable principal y otros archivos de biblioteca con una sola llamada API, en lugar de usar la API de almacenamiento para cargar archivos por separado, no se necesita más token de almacenamiento para administrar archivos.

Prerrequisitos

• Se requiere un token de Microsoft Entra para acceder a la API REST de Fabric. Se recomienda la biblioteca MSAL (Biblioteca de autenticación de Microsoft) para obtener el token. Para obtener más información, consulte Compatibilidad con el flujo de autenticación en MSAL.

La API Fabric REST define un punto final unificado para las operaciones CRUD de los elementos Fabric. El extremo es https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items.

Descripción del trabajo Spark v2 resumen del formato

En la carga útil de gestionar un elemento de definición de trabajo de Spark, el definition campo se utiliza para especificar la configuración detallada del elemento de definición de trabajo de Spark. El definition campo contiene dos subcampos: format y parts. El format campo especifica el formato del elemento de definición de trabajo de Spark, que debe ser SparkJobDefinitionV2 para el formato v2.

El parts campo es un array que contiene la configuración detallada del elemento de definición del trabajo de Spark. Cada elemento de la parts matriz representa una parte de la configuración detallada. Cada parte contiene tres subcampos: path, payloady payloadType. El path campo especifica la ruta de acceso de la parte, el payload campo especifica el contenido de la parte codificada en base64 y el payloadType campo especifica el tipo de la carga, que debe ser InlineBase64.

Crea un elemento de definición de trabajo en Spark con el archivo principal de definición y otros archivos de liberación

En el siguiente ejemplo, crearemos un elemento de definición de trabajo en Spark que:

1. El nombre es SJDHelloWorld.
2. El archivo principal de definición es main.py, que consiste en leer un archivo CSV desde su casa de lago predeterminada y guardarlo como tabla Delta de vuelta en la misma casa de lago.
3. Otro archivo lib es libs.py, que tiene una función de utilidad para devolver el nombre del archivo CSV y la tabla Delta.
4. La casa del lago por defecto está configurada con un ID de elemento específico de la casa del lago.

A continuación se muestra la carga útil detallada para crear el elemento de definición del trabajo de 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 descodificar o codificar la configuración detallada, puede usar las siguientes funciones auxiliares en Python. También hay otras herramientas en línea, como https://www.base64decode.org/ que pueden realizar el mismo trabajo.

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

Una respuesta HTTP code 202 indica que el elemento de definición de trabajo de Spark se creó con éxito.

Obtén definición de trabajo en Spark con partes de definición bajo el formato v2

Con el nuevo formato v2, al obtener un elemento de definición de trabajo de Spark con partes de definición, el contenido del archivo principal de definición y otros archivos lib se incluyen en la carga útil de respuesta, codificada en base64 bajo el parts campo. Aquí tienes un ejemplo de cómo obtener un elemento de definición de trabajo en Spark con partes de definición:

1. En primer lugar, realice una solicitud POST al punto de conexión https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{sjditemid}/getDefinitionParts?format=SparkJobDefinitionV2. Asegúrese de que el valor del parámetro de consulta de formato es SparkJobDefinitionV2.
2. A continuación, en los encabezados de respuesta, compruebe el código de estado HTTP. Un código HTTP 202 indica que la solicitud se aceptó correctamente. Copie el x-ms-operation-id valor de los encabezados de respuesta.
3. Por último, realice una solicitud GET al punto de conexión https://api.fabric.microsoft.com/v1/operations/{operationId} con el valor copiado x-ms-operation-id para obtener el resultado de la operación. En la carga útil de respuesta, el definition campo contiene la configuración detallada del elemento de definición de trabajo de Spark, incluyendo el archivo principal de definición y otros archivos de lib bajo el parts campo.

Actualiza el elemento de definición de trabajo de Spark con el archivo principal de definición y otros archivos de lib bajo el formato v2

Para actualizar un elemento de definición de trabajo de Spark existente con el archivo principal de definición y otros archivos de lib bajo el formato v2, puedes usar una estructura de carga útil similar a la operación de creación. Aquí tienes un ejemplo de actualización del elemento de definición de trabajo de Spark creado en la sección 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"
      }
    ]
  }
}

Con la carga anterior, se realizan los siguientes cambios en los archivos:

1. El archivo main.py se actualiza con nuevo contenido.
2. El lib1.py se elimina de este elemento de definición de trabajo en Spark y también se elimina del almacenamiento de OneLake.
3. Se añade un nuevo archivo de lib2.py a este elemento de definición de trabajo de Spark y se sube al almacenamiento de OneLake.

Para actualizar el elemento de definición de trabajo de Spark, realiza una solicitud POST al endpoint https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{sjditemid} con la carga útil mencionada anteriormente. Una respuesta con código HTTP 202 indica que el elemento de definición de trabajo de Spark se actualizó con éxito.

Contenido relacionado

• Programación y ejecución de una definición de trabajo de Apache Spark