Hvordan man opretter og opdaterer en Spark-jobdefinition med V2-format via Microsoft Fabric REST API

Spark-jobdefinition (SJD) er en type Fabric-produkt, der gør det muligt for brugere at definere og køre Apache Spark-jobs i Fabric. Spark-jobdefinitions-API'en v2 tillader brugere at oprette og opdatere Spark-jobdefinitionselementer med et nyt format kaldet SparkJobDefinitionV2. Den primære fordel ved at bruge v2-formatet er, at det tillader brugere at administrere hovedfilen og andre biblioteksfiler med ét enkelt API-kald; i stedet for at bruge storage API til at uploade filer separat, er der ikke behov for flere storage token til at håndtere filer.

Forudsætninger

• Et Microsoft Entra-token er nødvendigt for at få adgang til Fabric REST API'en. MSAL (Microsoft Authentication Library) biblioteket anbefales for at få tokenet. For mere information, se Authentication flow support i MSAL.

Fabric REST API'en definerer et samlet endepunkt for CRUD-operationer af Fabric-elementer. Slutpunktet er https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items.

Oversigt over Spark-jobdefinition v2 format

I payloaden til håndtering af et Spark-jobdefinitionselement definition bruges feltet til at specificere den detaljerede opsætning af Spark-jobdefinitionselementet. Feltet definition indeholder to delfelter: format og parts. Feltet format angiver formatet for Spark-jobdefinitionselementet, som skal være SparkJobDefinitionV2 for v2-formatet.

Feltet parts er et array, der indeholder den detaljerede opsætning af Spark-jobdefinitionselementet. Hvert element i arrayet repræsenterer parts en del af den detaljerede opsætning. Hver del indeholder tre underfelter: path, payload, og payloadType. Feltet path angiver delens sti, payload feltet angiver indholdet af den del, der er base64-kodet, og payloadType feltet angiver typen af nyttelasten, som skal være InlineBase64.

Opret et Spark-jobdefinitionselement med hoveddefinitionsfilen og andre bibliotekfiler

I det følgende eksempel opretter vi et Spark-jobdefinitionselement, som:

1. Navnet er SJDHelloWorld.
2. Hoveddefinitionsfilen er main.py, hvilket er at læse en CSV-fil fra dens standard lakehouse og gemme som en Delta-tabel tilbage til det samme lakehouse.
3. En anden lib-fil er libs.py, som har en hjælpefunktion til at returnere navnet på CSV-filen og Delta-tabellen.
4. Standard lakehouse er sat til et specifikt lakehouse item ID.

Følgende er den detaljerede payload til oprettelse af Spark-jobdefinitionselementet.

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

For at afkode eller kode den detaljerede opsætning kan du bruge følgende hjælpefunktioner i Python. Der findes også andre online værktøjer, som https://www.base64decode.org/ kan udføre det samme arbejde.

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

Et HTTP-kode 202-svar indikerer, at Spark-jobdefinitionselementet blev oprettet med succes.

Få Spark-jobdefinition med definitionsdele under v2-formatet

Med det nye v2-format, når man henter et Spark-jobdefinitionselement med definitionsdele, inkluderes filindholdet i hoveddefinitionsfilen og andre bibliotekfiler alle i responspayloaden, base64 kodet under feltet parts . Her er et eksempel på at få et Spark-jobdefinitionselement med definitionsdele:

1. Først laver du en POST-anmodning til endpointet https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{sjditemid}/getDefinitionParts?format=SparkJobDefinitionV2. Sørg for, at værdien af formatforespørgselsparameteren er SparkJobDefinitionV2.
2. Derefter tjekker du HTTP-statuskoden i svarheaderne. En HTTP-kode 202 angiver, at anmodningen blev accepteret med succes. Kopier værdien x-ms-operation-id fra svarheaderne.
3. Til sidst laver du en GET-anmodning til endepunktet https://api.fabric.microsoft.com/v1/operations/{operationId} med den kopierede x-ms-operation-id værdi for at få operationsresultatet. I responspayloaden definition indeholder feltet den detaljerede opsætning af Spark-jobdefinitionselementet, inklusive hoveddefinitionsfilen og andre bibliotekfiler under feltet parts .

Opdater Spark-jobdefinitionselementet med hoveddefinitionsfilen og andre bibliotekfiler under v2-format

For at opdatere et eksisterende Spark-jobdefinitionselement med hoveddefinitionsfilen og andre bibliotekfiler under v2-formatet, kan du bruge en lignende payload-struktur som oprettelsesoperationen. Her er et eksempel på opdatering af Spark-jobdefinitionselementet, som blev oprettet i det forrige afsnit:

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

Med ovenstående payload foretages følgende ændringer i filerne:

1. Den main.py fil opdateres med nyt indhold.
2. lib1.py er slettet fra dette Spark-jobdefinitionselement og også fjernet fra OneLake-lageret.
3. En ny lib2.py-fil tilføjes til dette Spark-jobdefinitionselement og uploades til OneLake-lageret.

For at opdatere Spark-jobdefinitionspunktet skal du lave en POST-anmodning til endepunktet https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{sjditemid} med ovenstående payload. Et HTTP-kode 202-svar indikerer, at Spark-jobdefinitionselementet blev opdateret korrekt.

Relateret indhold

• Planlæg og kør en Apache Spark-jobdefinition