Hvordan lage og oppdatere en Spark-jobbdefinisjon med V2-format via Microsoft Fabric REST API

Spark-jobbdefinisjon (SJD) er en type Fabric-element som lar brukere definere og kjøre Apache Spark-jobber i Fabric. Spark-jobbdefinisjons-API-et v2 lar brukere opprette og oppdatere Spark-jobbdefinisjonselementer med et nytt format kalt SparkJobDefinitionV2. Den viktigste fordelen med å bruke v2-formatet er at det lar brukere administrere hovedfilen og andre bibliotekfiler med ett enkelt API-kall, i stedet for å bruke lagrings-API for å laste opp filer separat, og det trengs ikke flere lagringstoken for å administrere filer.

Forutsetninger

• En Microsoft Entra-token kreves for å få tilgang til Fabric REST API. MSAL (Microsoft Authentication Library)-biblioteket anbefales for å få tak i tokenet. For mer informasjon, se Autentiseringsflytstøtte i MSAL.

Fabric REST API definerer et samlet endepunkt for CRUD-operasjoner av Fabric-elementer. Endepunktet er https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items.

Oversikt over Spark-jobbdefinisjon v2 format

I nyttelasten for å administrere et Spark-jobbdefinisjonselement definition brukes feltet til å spesifisere den detaljerte oppsettet av Spark-jobbdefinisjonselementet. Feltet definition inneholder to delfelt: format og parts. Feltet format spesifiserer formatet for Spark-jobbdefinisjonselementet, som skal være SparkJobDefinitionV2 for v2-formatet.

Feltet parts er et array som inneholder detaljert oppsett av Spark-jobbdefinisjonselementet. Hvert element i arrayet parts representerer en del av det detaljerte oppsettet. Hver del inneholder tre delfelt: path, payload, og payloadType. Feltet path spesifiserer banen til delen, payload feltet angir innholdet i delen som er base64-kodet, og payloadType feltet angir typen nyttelast, som skal være InlineBase64.

Lag et Spark-jobbdefinisjonselement med hoveddefinisjonsfilen og andre bibliotekfiler

I det følgende eksempelet lager vi et Spark-jobbdefinisjonselement som:

1. Navnet er SJDHelloWorld.
2. Hoveddefinisjonsfilen er main.py, som er å lese en CSV-fil fra standard lakehouse og lagre som en Delta-tabell tilbake til samme lakehouse.
3. En annen bibliotekfil er libs.py, som har en nyttefunksjon for å returnere navnet på CSV-filen og Delta-tabellen.
4. Standard lakehouse er satt til en spesifikk lakehouse-item-ID.

Følgende er den detaljerte nyttelasten for å lage Spark-jobbdefinisjonselementet.

{
  "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 å dekode eller kode det detaljerte oppsettet, kan du bruke følgende hjelpefunksjoner i Python. Det finnes også andre nettbaserte verktøy https://www.base64decode.org/ som kan utføre samme oppgave.

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-jobbdefinisjonselementet ble opprettet med suksess.

Få Spark-jobbdefinisjon med definisjonsdeler under v2-format

Med det nye v2-formatet, når man henter et Spark-jobbdefinisjonselement med definisjonsdeler, inkluderes filinnholdet i hoveddefinisjonsfilen og andre bibliotekfiler i responspayloaden, base64 kodet under feltet parts . Her er et eksempel på å få et Spark-jobbdefinisjonselement med definisjonsdeler:

1. Først, gjør en POST-forespørsel til endepunktet https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{sjditemid}/getDefinitionParts?format=SparkJobDefinitionV2. Sørg for at verdien til formatspørringsparameteren er SparkJobDefinitionV2.
2. Deretter sjekker du HTTP-statuskoden i svarhodene. En HTTP-kode 202 indikerer at forespørselen ble vellykket akseptert. Kopier x-ms-operation-id verdien fra svarhodene.
3. Til slutt gjør du en GET-forespørsel til endepunktet https://api.fabric.microsoft.com/v1/operations/{operationId} med den kopierte x-ms-operation-id verdien for å få operasjonsresultatet. I responsnyttelasten definition inneholder feltet den detaljerte oppsettet av Spark-jobbdefinisjonselementet, inkludert hoveddefinisjonsfilen og andre bibliotekfiler under parts feltet.

Oppdater Spark-jobbdefinisjonselementet med hoveddefinisjonsfilen og andre bibliotekfiler under v2-format

For å oppdatere et eksisterende Spark-jobbdefinisjonselement med hoveddefinisjonsfilen og andre bibliotekfiler under v2-formatet, kan du bruke en lignende nyttelaststruktur som opprettingsoperasjonen. Her er et eksempel på oppdatering av Spark-jobbdefinisjonselementet opprettet i forrige avsnitt:

{
  "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 den ovennevnte nyttelasten gjøres følgende endringer i filene:

1. main.py-filen oppdateres med nytt innhold.
2. lib1.py er slettet fra dette Spark-jobbdefinisjonselementet og også fjernet fra OneLake-lagringen.
3. En ny lib2.py-fil legges til dette Spark-jobbdefinisjonselementet og lastes opp til OneLake-lagringen.

For å oppdatere Spark-jobbdefinisjonselementet, gjør en POST-forespørsel til endepunktet https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{sjditemid} med den ovennevnte nyttelasten. Et HTTP-kode 202-svar indikerer at Spark-jobbdefinisjonselementet ble oppdatert med suksess.

Beslektet innhold

• Planlegg og kjør en Apache Spark-jobbdefinisjon