Microsoft Spark Utilities (MSSparkUtils) for Fabric

Microsoft Spark Utilities (MSSparkUtils) er en innebygd pakke som hjelper deg enkelt å utføre vanlige oppgaver. Bruk MSSparkUtils til å arbeide med filsystemer, få miljøvariabler, kjede notatbøker sammen og arbeide med hemmeligheter. MSSparkUtils-pakken er tilgjengelig i PySpark (Python), Scala, SparkR notebooks og Fabric-pipelines.

Filsystemverktøy

mssparkutils.fs tilbyr verktøy for arbeid med ulike filsystemer, inkludert Azure Data Lake Storage Gen2 og Azure Blob Storage. Kontroller at du konfigurerer tilgang til Azure Data Lake Storage Gen2 og Azure Blob Storage på riktig måte.

Kjør følgende kommandoer for en oversikt over de tilgjengelige metodene:

from notebookutils import mssparkutils
mssparkutils.fs.help()

Utdata

mssparkutils.fs provides utilities for working with various FileSystems.

Below is overview about the available methods:

cp(from: String, to: String, recurse: Boolean = false): Boolean -> Copies a file or directory, possibly across FileSystems
mv(from: String, to: String, recurse: Boolean = false): Boolean -> Moves a file or directory, possibly across FileSystems
ls(dir: String): Array -> Lists the contents of a directory
mkdirs(dir: String): Boolean -> Creates the given directory if it does not exist, also creating any necessary parent directories
put(file: String, contents: String, overwrite: Boolean = false): Boolean -> Writes the given String out to a file, encoded in UTF-8
head(file: String, maxBytes: int = 1024 * 100): String -> Returns up to the first 'maxBytes' bytes of the given file as a String encoded in UTF-8
append(file: String, content: String, createFileIfNotExists: Boolean): Boolean -> Append the content to a file
rm(dir: String, recurse: Boolean = false): Boolean -> Removes a file or directory
exists(file: String): Boolean -> Check if a file or directory exists
mount(source: String, mountPoint: String, extraConfigs: Map[String, Any]): Boolean -> Mounts the given remote storage directory at the given mount point
unmount(mountPoint: String): Boolean -> Deletes a mount point
mounts(): Array[MountPointInfo] -> Show information about what is mounted
getMountPath(mountPoint: String, scope: String = ""): String -> Gets the local path of the mount point

Use mssparkutils.fs.help("methodName") for more info about a method.

MSSparkUtils fungerer med filsystemet på samme måte som Spark API-er. Ta for eksempel mssparkuitls.fs.mkdirs() og lakehouse-bruk:

Listefiler

Hvis du vil vise innholdet i en katalog, bruker du mssparkutils.fs.ls('Katalogbanen'). Eksempel:

mssparkutils.fs.ls("Files/tmp") # works with the default lakehouse files using relative path 
mssparkutils.fs.ls("abfss://<container_name>@<storage_account_name>.dfs.core.windows.net/<path>")  # based on ABFS file system 
mssparkutils.fs.ls("file:/tmp")  # based on local file system of driver node 

Vis filegenskaper

Denne metoden returnerer filegenskaper, inkludert filnavn, filsti, filstørrelse og om det er en mappe eller en fil.

files = mssparkutils.fs.ls('Your directory path')
for file in files:
    print(file.name, file.isDir, file.isFile, file.path, file.size)

Opprett ny katalog

Denne metoden oppretter den spesifiserte katalogen hvis den ikke eksisterer, og oppretter nødvendige overordnede kataloger.

mssparkutils.fs.mkdirs('new directory name')  
mssparkutils.fs. mkdirs("Files/<new_dir>")  # works with the default lakehouse files using relative path 
mssparkutils.fs.ls("abfss://<container_name>@<storage_account_name>.dfs.core.windows.net/<new_dir>")  # based on ABFS file system 
mssparkutils.fs.ls("file:/<new_dir>")  # based on local file system of driver node 

Kopier fil

Denne metoden kopierer en fil eller katalog, og støtter kopieringsaktivitet på tvers av filsystemer.

mssparkutils.fs.cp('source file or directory', 'destination file or directory', True)# Set the third parameter as True to copy all files and directories recursively

Performant kopieringsfil

Denne metoden gir en raskere måte å kopiere eller flytte filer på, spesielt store mengder data.

mssparkutils.fs.fastcp('source file or directory', 'destination file or directory', True)# Set the third parameter as True to copy all files and directories recursively

Forhåndsvis filinnhold

Denne metoden returnerer opptil de første maxBytes bytene av den angitte filen som en streng kodet i UTF-8.

# Set the second parameter as an integer for the maxBytes to read
mssparkutils.fs.head('file path', <maxBytes>)

Flytt fil

Denne metoden flytter en fil eller katalog, og støtter flyttinger på tvers av filsystemer.

mssparkutils.fs.mv('source file or directory', 'destination directory', True) # Set the last parameter as True to firstly create the parent directory if it does not exist
mssparkutils.fs.mv('source file or directory', 'destination directory', True, True) # Set the third parameter to True to firstly create the parent directory if it does not exist. Set the last parameter to True to overwrite the updates.

Skrive fil

Denne metoden skriver den angitte strengen ut til en fil, kodet i UTF-8.

mssparkutils.fs.put("file path", "content to write", True) # Set the last parameter as True to overwrite the file if it existed already

Tilføy innhold til en fil

Denne metoden tilføyer den angitte strengen til en fil, kodet i UTF-8.

mssparkutils.fs.append("file path", "content to append", True) # Set the last parameter as True to create the file if it does not exist

Slette fil eller katalog

Denne metoden fjerner en fil eller katalog.

mssparkutils.fs.rm('file path', True) # Set the last parameter as True to remove all files and directories recursively

Monter/demonter katalog

For mer informasjon om detaljert bruk, se Filmontering og avmontering.

Notatblokkverktøy

Bruk MSSparkUtils Notebook Utilities til å kjøre en notatblokk eller avslutte en notatblokk med en verdi. Kjør følgende kommando for å få en oversikt over de tilgjengelige metodene:

mssparkutils.notebook.help()

Ytelse:

exit(value: String): Raises NotebookExit Exception -> This method lets you exit a notebook with a value.
run(path: String, timeoutSeconds: int, arguments: Map): String -> This method runs a notebook and returns its exit value.

Referere til en notatblokk

Denne metoden refererer til en notatblokk og returnerer avslutningsverdien. Du kan kjøre nestefunksjonskall i en notatblokk interaktivt eller i et datasamlebånd. Notatblokken som det refereres til, kjøres i Spark-utvalget i notatblokken som kaller denne funksjonen.

mssparkutils.notebook.run("notebook name", <timeoutSeconds>, <parameterMap>, <workspaceId>)

Eksempel:

mssparkutils.notebook.run("Sample1", 90, {"input": 20 })

Stoffnotatblokk støtter også referanse til notatblokker på tvers av flere arbeidsområder ved å angi arbeidsområdets ID.

mssparkutils.notebook.run("Sample1", 90, {"input": 20 }, "fe0a6e2a-a909-4aa3-a698-0a651de790aa")

Du kan åpne øyeblikksbildekoblingen for referansekjøringen i celleutdataene. Øyeblikksbildet registrerer resultatene for kjøring av kode og lar deg enkelt feilsøke en referansekjøring.

Referanse kjøre flere notatblokker parallelt

Med metoden mssparkutils.notebook.runMultiple() kan du kjøre flere notatblokker parallelt eller med en forhåndsdefinert topologisk struktur. API-et bruker en multitrådet implementering for å sende inn, køe og overvåke barnenotatbøker som kjører på isolerte REPL-instanser (lese-eval-print-loop) innenfor den eksisterende Spark-økten. De refererte barnenotatbøkene deler sesjonens beregningsressurser.

Med mssparkutils.notebook.runMultiple()kan du:

• Utfør flere notatblokker samtidig, uten å vente på at hver av dem skal fullføres.

• Angi avhengighetene og rekkefølgen på kjøringen for notatblokkene, ved hjelp av et enkelt JSON-format.

• Optimaliser bruken av Spark-databehandlingsressurser og reduser kostnadene for Fabric-prosjektene dine.

• Se øyeblikksbildene av hver notatbokkjøringspost i utgangen, og feilsøk og overvåk notatblokkoppgavene dine på en praktisk måte.

• Hent avslutningsverdien for hver lederaktivitet, og bruk dem i nedstrømsoppgaver.

Du kan også prøve å kjøre mssparkutils.notebook.help("runMultiple") for å finne eksemplet og detaljert bruk.

Her er et enkelt eksempel på hvordan du kjører en liste over notatblokker parallelt ved hjelp av denne metoden:

mssparkutils.notebook.runMultiple(["NotebookSimple", "NotebookSimple2"])

Utføringsresultatet fra rotnotatblokken er som følger:

Følgende eksempel viser kjøring av notatbøker med en topologisk struktur ved å bruke mssparkutils.notebook.runMultiple(). Bruk denne metoden til enkelt å organisere notatblokker gjennom en kodeopplevelse.

# run multiple notebooks with parameters
DAG = {
    "activities": [
        {
            "name": "NotebookSimple", # activity name, must be unique
            "path": "NotebookSimple", # notebook path
            "timeoutPerCellInSeconds": 90, # max timeout for each cell, default to 90 seconds
            "args": {"p1": "changed value", "p2": 100}, # notebook parameters
        },
        {
            "name": "NotebookSimple2",
            "path": "NotebookSimple2",
            "timeoutPerCellInSeconds": 120,
            "args": {"p1": "changed value 2", "p2": 200}
        },
        {
            "name": "NotebookSimple2.2",
            "path": "NotebookSimple2",
            "timeoutPerCellInSeconds": 120,
            "args": {"p1": "changed value 3", "p2": 300},
            "retry": 1,
            "retryIntervalInSeconds": 10,
            "dependencies": ["NotebookSimple"] # list of activity names that this activity depends on
        }
    ],
    "timeoutInSeconds": 43200, # max timeout for the entire DAG, default to 12 hours
    "concurrency": 50 # max number of notebooks to run concurrently, defaults to 50 but ultimately constrained by the number of driver cores
}
mssparkutils.notebook.runMultiple(DAG, {"displayDAGViaGraphviz": False})

Utføringsresultatet fra rotnotatblokken er som følger:

Avslutte en notatblokk

Denne metoden avslutter en notatblokk med en verdi. Du kan kjøre nestefunksjonskall i en notatblokk interaktivt eller i et datasamlebånd.

• Når du kaller en exit() -funksjon fra en notatblokk interaktivt, kaster Fabric-notatblokken et unntak, hopper over etterfølgende celler og holder Spark-økten i live.

• Når du orkestrerer en notatblokk i et datasamlebånd som kaller en exit() -funksjon, returnerer notatblokkaktiviteten med en avslutningsverdi, fullfører datasamlebåndkjøringen og stopper Spark-økten. Ikke legg exit() -funksjonen rundt en try/catch, da denne NotebookExit-unntaket må propagere for at pipelinen skal få returverdien.

• Når du kaller en exit()-funksjon i en notatbok som refereres, stopper Fabric Spark videre kjøring av den refererte notatboken, og fortsetter å kjøre de neste cellene i hovednotatboken som kaller run()-funksjonen. Eksempel: Notatblokk1 har tre celler og kaller en exit()-funksjon i den andre cellen. Notatblokk2 har fem celler og kaller kjøring (notatblokk1) i den tredje cellen. Når du kjører Notatblokk2, stopper Notatblokk1 i den andre cellen når du trykker på exit() -funksjonen. Notatblokk2 fortsetter å kjøre sin fjerde celle og femte celle.

mssparkutils.notebook.exit("value string")

Eksempel:

Eksempel på 1 notatblokk med følgende to celler:

• Celle 1 definerer en inndataparameter med standardverdien satt til 10.

• Celle 2 avslutter notatblokken med inndata som avslutningsverdi.

Du kan kjøre Eksempel1 i en annen notatblokk med standardverdier:

exitVal = mssparkutils.notebook.run("Sample1")
print (exitVal)

Ytelse:

Notebook executed successfully with exit value 10

Du kan kjøre eksempel1 i en annen notatblokk og angi inndataverdien som 20:

exitVal = mssparkutils.notebook.run("Sample1", 90, {"input": 20 })
print (exitVal)

Ytelse:

Notebook executed successfully with exit value 20

Legitimasjonsverktøy

Du kan bruke MSSparkUtils Credentials Utilities for å få tilgang til tokens og administrere hemmeligheter i Azure Key Vault.

Kjør følgende kommando for å få en oversikt over de tilgjengelige metodene:

mssparkutils.credentials.help()

Ytelse:

getToken(audience, name): returns AAD token for a given audience, name (optional)
getSecret(keyvault_endpoint, secret_name): returns secret for a given Key Vault and secret name

Hent token

getTokenreturnerer et Microsoft Entra-token for et gitt publikum og navn (valgfritt). Listen nedenfor viser de tilgjengelige målgruppenøklene:

• Lagringsressurs: storage
• Power BI-ressurs:pbi
• Azure Key Vault Resource:keyvault
• Synapse RTA KQL DB Ressurs:kusto

Kjør følgende kommando for å hente tokenet:

mssparkutils.credentials.getToken('audience Key')

Få hemmeligheter ved å bruke brukerlegitimasjon

getSecretreturnerer en Azure Key Vault-hemmelighet for et gitt Azure Key Vault-endepunkt og hemmelig navn ved å bruke brukerlegitimasjon.

mssparkutils.credentials.getSecret('https://<name>.vault.azure.net/', 'secret name')

Filmontering og demontering

Fabric støtter følgende monteringsscenarioer i Microsoft Spark Utilities-pakken. Du kan bruke API-ene mount, unmount, getMountPath() og mounts() for å koble ekstern lagring (Azure Data Lake Storage Gen2) til alle fungerende noder (drivernode og arbeider-noder). Når lagringsmonteringspunktet er på plass, kan du bruke den lokale fil-API-en til å få tilgang til data som om den er lagret i det lokale filsystemet.

How to mount an Azure Data Lake Storage Gen2-konto

Følgende eksempel viser hvordan man monterer Azure Data Lake Storage Gen2. Montering av Blob Storage fungerer på samme måte.

Dette eksemplet forutsetter at du har én Data Lake Storage Gen2-konto med navnet Storegen2, og kontoen har én beholder med navnet mycontainer som du vil montere til /test i spark-økten for notatblokken.

For å montere containeren som heter mycontainer, sjekker mssparkutils først om du har tillatelse til å få tilgang til containeren. Fabric støtter tre autentiseringsmetoder for trigger-mount-operasjonen: Microsoft Entra-token (standard og anbefalt), accountKey og sastoken. For mer informasjon om Microsoft Entra-tokenautentisering og dagens notebookutils API, se NotebookUtils file mount and unmount for Fabric.

Monter ved å bruke en delt tilgangssignatur eller kontonøkkel

MSSparkUtils støtter eksplisitt å sende en kontonøkkel eller SAS-token (Shared Access Signature) som en parameter for å montere målet.

Av sikkerhetsgrunner anbefaler vi at du lagrer kontonøkler eller SAS-tokener i Azure Key Vault (som følgende skjermbilde viser). Deretter kan du hente dem ved hjelp av mssparkutils.credentials.getSecret API. Hvis du vil ha mer informasjon om Azure Key Vault, kan du se Om administrerte lagringskontonøkler for Azure Key Vault.

Eksempelkode for accountKey-metoden :

from notebookutils import mssparkutils  
# get access token for keyvault resource
# you can also use full audience here like https://vault.azure.net
accountKey = mssparkutils.credentials.getSecret("<vaultURI>", "<secretName>")
mssparkutils.fs.mount(  
    "abfss://mycontainer@<accountname>.dfs.core.windows.net",  
    "/test",  
    {"accountKey":accountKey}
)

Eksempelkode for sastoken:

from notebookutils import mssparkutils  
# get access token for keyvault resource
# you can also use full audience here like https://vault.azure.net
sasToken = mssparkutils.credentials.getSecret("<vaultURI>", "<secretName>")
mssparkutils.fs.mount(  
    "abfss://mycontainer@<accountname>.dfs.core.windows.net",  
    "/test",  
    {"sasToken":sasToken}
)

from notebookutils import mssparkutils

Monteringsparametere:

• fileCacheTimeout: Blobs cacher i den lokale temp-mappen i 120 sekunder som standard. I denne perioden sjekker ikke blobfuse om filen er oppdatert. Sett denne parameteren til å endre standard timeout. Når flere klienter endrer filer samtidig, for å unngå inkonsistenser mellom lokale og eksterne filer, anbefaler vi at du forkorter cache-tiden, eller til og med endrer den til 0, og alltid henter de nyeste filene fra serveren.
• timeout: Mount-operasjonens timeout er som standard 120 sekunder. Sett denne parameteren til å endre standard timeout. Når det er for mange eksekutorer eller når monteringen går ut på tid, anbefaler vi at du øker verdien.

Du kan bruke disse parameterne slik:

mssparkutils.fs.mount(
   "abfss://mycontainer@<accountname>.dfs.core.windows.net",
   "/test",
   {"fileCacheTimeout": 120, "timeout": 120}
)

Slik monterer du et lakehouse

Eksempelkode for montering av et hytte ved innsjøen /test:

from notebookutils import mssparkutils 
mssparkutils.fs.mount( 
 "abfss://<workspace_id>@onelake.dfs.fabric.microsoft.com/<lakehouse_id>", 
 "/test"
)

Få tilgang til filer under monteringspunktet ved å bruke mssparkutils fs API-et

Hovedformålet med mount-operasjonen er å la deg få tilgang til dataene lagret i en ekstern lagringskonto ved å bruke et lokalt filsystem-API. Du kan også få tilgang til dataene ved hjelp av mssparkutils fs API med en montert bane som parameter. Dette baneformatet er litt annerledes.

Anta at du monterte Data Lake Storage Gen2-containeren mycontainer til /test ved å bruke mount-API-et. Når du får tilgang til dataene ved å bruke et lokalt filsystem-API, er stiformatet slik:

/synfs/notebook/{sessionId}/test/{filename}

Når du ønsker å få tilgang til dataene ved å bruke mssparkutils fs API, anbefaler vi at du bruker getMountPath() for å få den nøyaktige stien:

path = mssparkutils.fs.getMountPath("/test")

• Listekataloger:

  mssparkutils.fs.ls(f"file://{mssparkutils.fs.getMountPath('/test')}")
  

• Les filinnhold:

  mssparkutils.fs.head(f"file://{mssparkutils.fs.getMountPath('/test')}/myFile.txt")
  

• Opprette en katalog:

  mssparkutils.fs.mkdirs(f"file://{mssparkutils.fs.getMountPath('/test')}/newdir")
  

Få tilgang til filer under monteringspunktet via lokal bane

Du kan enkelt lese og skrive filene i monteringspunktet ved hjelp av standard filsystem. Her er et Python-eksempel:

#File read
with open(mssparkutils.fs.getMountPath('/test2') + "/myFile.txt", "r") as f:
    print(f.read())
#File write
with open(mssparkutils.fs.getMountPath('/test2') + "/myFile.txt", "w") as f:
    print(f.write("dummy data"))

Slik kontrollerer du eksisterende monteringspunkter

Du kan bruke mssparkutils.fs.mounts() API til å kontrollere all eksisterende informasjon om monteringspunkt:

mssparkutils.fs.mounts()

Slik fjerner du monteringspunktet

Bruk følgende kode til å demontere monteringspunktet (/test i dette eksemplet):

mssparkutils.fs.unmount("/test")

Kjente begrensninger

• Den nåværende monteringen er en jobbnivå-konfigurasjon. Vi anbefaler at du bruker mounts-API-et for å sjekke om et monteringspunkt finnes eller ikke er tilgjengelig.

• Demonteringsmekanismen er ikke automatisk. Når programmet kjøres, må du eksplisitt kalle opp en umontert API i koden for å demontere monteringspunktet og frigjøre diskplassen. Ellers eksisterer monteringspunktet fortsatt i noden etter at applikasjonskjøringen er ferdig.

• Montering av en Azure Data Lake Storage Gen1-lagringskonto støttes ikke.

Lakehouse verktøy

Modulen mssparkutils.lakehouse gir verktøy for håndtering av innsjøhus. Disse verktøyene gjør det enkelt å opprette, hente ut, oppdatere og slette elementer i innsjøhuset.

Oversikt over metoder

Følgende metoder er tilgjengelige i modulen mssparkutils.lakehouse :

# Create a new Lakehouse artifact
create(name: String, description: String = "", workspaceId: String = ""): Artifact

# Retrieve a Lakehouse artifact
get(name: String, workspaceId: String = ""): Artifact

# Update an existing Lakehouse artifact
update(name: String, newName: String, description: String = "", workspaceId: String = ""): Artifact

# Delete a Lakehouse artifact
delete(name: String, workspaceId: String = ""): Boolean

# List all Lakehouse artifacts
list(workspaceId: String = ""): Array[Artifact]

Eksempler på bruk

For å bruke disse metodene effektivt, vurder følgende brukseksempler:

Å lage et innsjøhus-objekt

artifact = mssparkutils.lakehouse.create("artifact_name", "Description of the artifact", "optional_workspace_id")

Henting av en gjenstand i et innsjøhus

artifact = mssparkutils.lakehouse.get("artifact_name", "optional_workspace_id")

Oppdatering av et innsjøhus-element

updated_artifact = mssparkutils.lakehouse.update("old_name", "new_name", "Updated description", "optional_workspace_id")

Sletting av et innsjøhus-element

is_deleted = mssparkutils.lakehouse.delete("artifact_name", "optional_workspace_id")

Oversikt over gjenstander i innsjøhuset

artifacts_list = mssparkutils.lakehouse.list("optional_workspace_id")

Tilleggsinformasjon

For mer detaljert informasjon om hver metode og dens parametere, bruk funksjonen mssparkutils.lakehouse.help("methodName") .

Ved å bruke MSSparkUtils' Lakehouse-verktøy kan du mer effektivt administrere dine lakehouse-elementer og integrere denne administrasjonen i dine Fabric-pipelines, noe som forbedrer din totale databehandlingsopplevelse.

Utforsk disse verktøyene og integrer dem i dine Fabric-arbeidsflyter for sømløs lakehouse-varehåndtering.

Kjøretidsverktøy

Vis kontekstinformasjon for økten

Ved å bruke mssparkutils.runtime.context, kan du få kontekstinformasjon for den nåværende live-økten, inkludert notatbokens navn, standard lakehouse, arbeidsområdeinformasjon, om det er en pipeline-kjøring, og mer.

mssparkutils.runtime.context

Kjent problem

Når du bruker en runtime-versjon som er senere enn 1.2 og kjører mssparkutils.help(), støttes ikke de oppførte fabricClient-, warehouse- og workspace-API-ene for øyeblikket.

Relatert innhold

• Bibliotekbehandling