Microsoft Spark Utilities (MSSparkUtils) för Fabric

Microsoft Spark Utilities (MSSparkUtils) är ett inbyggt paket som hjälper dig att enkelt utföra vanliga uppgifter. Använd MSSparkUtils för att arbeta med filsystem, hämta miljövariabler, länka ihop notebook-filer och arbeta med hemligheter. MSSparkUtils-paketet finns tillgängligt i PySpark (Python), Scala, SparkR-notebooks och Fabric-pipelines.

Filsystemverktyg

mssparkutils.fs tillhandahåller verktyg för att arbeta med olika filsystem, inklusive Azure Data Lake Storage Gen2 och Azure Blob Storage. Se till att du konfigurerar åtkomsten till Azure Data Lake Storage Gen2 och Azure Blob Storage på rätt sätt.

Kör följande kommandon för en översikt över tillgängliga metoder:

from notebookutils import mssparkutils
mssparkutils.fs.help()

Resultat

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 fungerar med filsystemet på samma sätt som Spark-API:er. Ta mssparkuitls.fs.mkdirs() och lakehouse-användningen som exempel:

Visa filerna

Om du vill visa en lista över innehållet i en katalog använder du mssparkutils.fs.ls('Din katalogsökväg'). Till exempel:

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 

Visa filegenskaper

Denna metod returnerar filegenskaper, inklusive filnamn, filsökväg, filstorlek och om det är en katalog 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)

Skapa ny katalog

Denna metod skapar den angivna katalogen om den inte finns, och skapar nödvändiga föräldrakataloger.

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 

Kopiera fil

Den här metoden kopierar en fil eller katalog och stöder kopieringsaktivitet mellan filsystem.

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

Effektiv kopiering av fil

Den här metoden ger ett snabbare sätt att kopiera eller flytta filer, särskilt stora mängder 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

Förhandsgranska filinnehåll

Denna metod returnerar upp till de första maxBytes bytena av den angivna filen som en sträng kodad i UTF-8.

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

Flytta fil

Den här metoden flyttar en fil eller katalog och stöder flyttningar mellan filsystem.

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.

Skriv fil

Den här metoden skriver ut den angivna strängen till en fil, kodad 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

Lägga till innehåll i en fil

Den här metoden lägger till den angivna strängen i en fil, kodad 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

Ta bort fil eller katalog

Den här metoden tar bort en fil eller katalog.

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

Montera/demontera katalog

För mer information om mer detaljerad användning, se Montera och avmontera filer.

Verktyg för notebooks

Använd MSSparkUtils Notebook Utilities för att köra en notebook-fil eller avsluta en notebook-fil med ett värde. Kör följande kommando för att få en översikt över tillgängliga metoder:

mssparkutils.notebook.help()

Resultat:

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.

Referera till en anteckningsbok

Den här metoden refererar till en notebook-fil och returnerar dess slutvärde. Du kan köra kapslingsfunktionsanrop i en notebook-fil interaktivt eller i en pipeline. Notebook-filen som refereras körs i Spark-poolen i notebook-filen som anropar den här funktionen.

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

Till exempel:

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

Notebook-filen Fabric har också stöd för att referera till notebook-filer över flera arbetsytor genom att ange arbetsyte-ID .

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

Du kan öppna länken till ögonblicksbilden av referenskörningen i cellens utdata. Ögonblicksbilden fångar kodkörningsresultatet och gör att du enkelt kan felsöka en referenskörning.

Referenskörning av flera notebook-filer parallellt

Med metoden mssparkutils.notebook.runMultiple() kan du köra flera notebook-filer parallellt eller med en fördefinierad topologisk struktur. API:et använder en multitrådad implementation för att skicka in, köa och övervaka barnnotebooks som körs på isolerade REPL-instanser (read-eval-print-loop) inom den befintliga Spark-sessionen. De refererade barnanteckningsböckerna delar sessionens beräkningsresurser.

Med mssparkutils.notebook.runMultiple()kan du:

• Kör flera notebook-filer samtidigt, utan att vänta på att var och en ska slutföras.

• Ange beroenden och körningsordning för dina notebook-filer med hjälp av ett enkelt JSON-format.

• Optimera användningen av Spark-beräkningsresurser och minska kostnaden för dina Fabric-projekt.

• Se snapshots av varje anteckningsbok i utdata, och felsök och övervaka dina anteckningsboksuppgifter på ett smidigt sätt.

• Hämta slutvärdet för varje verkställande aktivitet och använd dem i underordnade uppgifter.

Du kan också försöka köra mssparkutils.notebook.help("runMultiple") för att hitta exemplet och den detaljerade användningen.

Här är ett enkelt exempel på hur du kör en lista över notebook-filer parallellt med den här metoden:

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

Körningsresultatet från rot-anteckningsboken är följande:

Följande exempel visar att man kör anteckningsböcker med en topologisk struktur genom att använda mssparkutils.notebook.runMultiple(). Använd den här metoden för att enkelt orkestrera notebook-filer via en kodupplevelse.

# 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})

Körningsresultatet från rot-anteckningsboken är följande:

Avsluta en notebook

Den här metoden avslutar en anteckningsbok med ett värde. Du kan köra kapslingsfunktionsanrop i en notebook-fil interaktivt eller i en pipeline.

• När du anropar en exit() -funktion från en notebook-fil interaktivt genererar Fabric-notebook-filen ett undantag, hoppar över efterföljande celler och håller Spark-sessionen vid liv.

• När du orkestrerar en notebook i en pipeline som anropar en exit()-funktion returneras notebook-aktiviteten med ett exitvärde, slutför pipeline-körningen och stoppar Spark-sessionen. Omslut inte funktionen exit() med try/catch, eftersom undantaget NotebookExit måste kunna propaggera för att pipelinen ska få returvärdet.

• När du anropar en exit()-funktion i en notebook som refereras, stoppar Fabric Spark den fortsatta exekveringen av den refererade notebooken och fortsätter att köra nästa celler i huvudnotebooken som anropar run()-funktionen. Till exempel: Notebook1 har tre celler och anropar en exit() -funktion i den andra cellen. Notebook2 har fem celler och anropar run(notebook1) i den tredje cellen. När du kör Notebook2 stoppas Notebook1 vid den andra cellen när du trycker på funktionen exit(). Notebook2 fortsätter att köra sin fjärde och femte cell.

mssparkutils.notebook.exit("value string")

Till exempel:

Sample1 Notebook med följande två celler:

• Cell 1 definierar en indataparameter med standardvärdet inställt på 10.

• Cell 2 lämnar notebooken med indata som utgångsvärde.

Du kan köra Sample1 i en annan notebook-fil med standardvärden:

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

Resultat:

Notebook executed successfully with exit value 10

Du kan köra Sample1 i en annan notebook-fil och ange indatavärdet som 20:

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

Resultat:

Notebook executed successfully with exit value 20

Verktyg för autentiseringsuppgifter

Du kan använda MSSparkUtils Credentials Utilities för att få åtkomsttokens och hantera hemligheter i Azure Key Vault.

Kör följande kommando för att få en översikt över tillgängliga metoder:

mssparkutils.credentials.help()

Resultat:

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

Hämta token

getTokenreturnerar en Microsoft Entra-token för en given målgrupp och namn (valfritt). I följande lista visas tillgängliga målgruppsnycklar:

• Lagringspublikresurs: storage
• Power BI-resurs:pbi
• Azure Key Vault Resource:keyvault
• Synapse RTA KQL DB Resurs: kusto

Kör följande kommando för att hämta token:

mssparkutils.credentials.getToken('audience Key')

Få hemlighet genom att använda användaruppgifter

getSecret returnerar en Azure Key Vault-hemlighet för en angiven Azure Key Vault-slutpunkt och ett hemlighetsnamn med hjälp av användarautentiseringsuppgifter.

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

Montera och demontera filer

Fabric stöder följande monteringsscenarier i Microsoft Spark Utilities-paketet. Du kan använda API:erna mount, unmount,getMountPath() och mounts() för att koppla fjärrlagring (Azure Data Lake Storage Gen2) till alla fungerande noder (drivernod och worker-noder). När lagringsmonteringspunkten är på plats använder du det lokala fil-API:et för att komma åt data som om de lagras i det lokala filsystemet.

Hur man monterar ett Azure Data Lake Storage Gen2-konto

Följande exempel visar hur man monterar Azure Data Lake Storage Gen2. Montering av Blob Storage fungerar på liknande sätt.

Det här exemplet förutsätter att du har ett Data Lake Storage Gen2-konto med namnet storegen2, och kontot har en container med namnet mycontainer som du vill montera till /test i din Notebook Spark-session.

För att montera containern med namnet mycontainer kontrollerar mssparkutils först om du har behörighet att komma åt containern. Fabric stöder tre autentiseringsmetoder för triggermonteringsoperationen: Microsoft Entra-token (standard och rekommenderad), accountKey och sastoken. För mer information om Microsoft Entra-tokenautentisering och det nuvarande notebookutils API:et, se Montering och avmontering av NotebookUtils-filer för Fabric.

Montera genom att använda en delad åtkomstsignaturtoken eller kontonyckel

MSSparkUtils stöder explicit överföring av en kontonyckel eller SAS-token (Signatur för delad åtkomst) som en parameter för att montera målet.

Av säkerhetsskäl rekommenderar vi att du lagrar kontonycklar eller SAS-token i Azure Key Vault (som följande skärmbild visar). Du kan sedan hämta dem med hjälp av API:et mssparkutils.credentials.getSecret . Mer information om Azure Key Vault finns i Om Azure Key Vault-hanterade lagringskontonycklar.

Exempelkod för metoden accountKey :

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}
)

Exempelkod för 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

Monteringsparametrar:

• fileCacheTimeout: Blobs lagras i cacheminnet i den lokala temporära mappen som standard i 120 sekunder. Under den här tiden kontrollerar blobfuse inte om filen är uppdaterad. Ställ in denna parameter för att ändra standardtimeouten. När flera klienter ändrar filer samtidigt, för att undvika inkonsekvenser mellan lokala och fjärrfiler, rekommenderar vi att du förkortar cachetiden, eller till och med ändrar den till 0, och alltid hämtar de senaste filerna från servern.
• timeout: Mount-operationens timeout är som standard 120 sekunder. Ställ in denna parameter för att ändra standardtimeouten. Om det finns för många exekverare eller om monteringen överskrider tidsgränsen rekommenderar vi att du ökar värdet.

Du kan använda följande parametrar:

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

Hur man monterar ett sjöhus

Exempelkod för att montera ett sjöhus på /test:

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

Åtkomst till filer under monteringspunkten genom att använda mssparkutils fs API

Huvudsyftet med mount-operationen är att låta dig komma åt data som lagras i ett fjärrlagringskonto via ett lokalt filsystem-API. Du kan också komma åt data med hjälp av mssparkutils fs API:t med en monterad sökväg som argument. Det här sökvägsformatet är lite annorlunda.

Anta att du monterade Data Lake Storage Gen2-containern mycontainer på /test med hjälp av monterings-API:t. När du kommer åt datan via ett lokalt filsystem-API är sökvägsformatet så här:

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

När du vill komma åt data via mssparkutils fs API rekommenderar vi att du använder getMountPath() för att få den korrekta vägen:

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

• Lista kataloger:

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

• Läsa filinnehåll:

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

• Skapa en katalog:

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

Komma åt filer under monteringspunkten via lokal sökväg

Du kan enkelt läsa och skriva filerna i monteringspunkten med hjälp av standardfilsystemet. Här är ett Python-exempel:

#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"))

Så här kontrollerar du befintliga monteringspunkter

Du kan använda API:et mssparkutils.fs.mounts() för att kontrollera all befintlig monteringspunktsinformation:

mssparkutils.fs.mounts()

Avmontera monteringspunkten

Använd följande kod för att demontera monteringspunkten (/testa i det här exemplet):

mssparkutils.fs.unmount("/test")

Kända begränsningar

• Den aktuella monteringen är en konfiguration på jobbnivå. Vi rekommenderar att du använder mounts-API :et för att kontrollera om en monteringspunkt finns eller inte är tillgänglig.

• Avmonteringsmekanismen är inte automatisk. När programkörningen är klar måste du uttryckligen anropa ett avmonterings-API i koden för att demontera monteringspunkten och frigöra diskutrymmet. Annars finns monteringspunkten fortfarande i noden när programkörningen har slutförts.

• Att montera ett Azure Data Lake Storage Gen1-lagringskonto stöds inte.

Lakehouse-verktyg

Modulen mssparkutils.lakehouse tillhandahåller verktyg för att hantera sjöhusföremål. Dessa verktyg gör det enkelt att skapa, hämta, uppdatera och ta bort sjöhusobjekt.

Översikt över metoder

Följande metoder finns tillgängliga 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]

Exempel på användning

För att använda dessa metoder effektivt, överväg följande användningsexempel:

Skapa ett sjöhusobjekt

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

Hämta ett sjöhusföremål

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

Uppdatera ett sjöhus-objekt

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

Raderar ett sjöhusobjekt

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

Lista över sjöhusobjekt

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

Ytterligare information

För mer detaljerad information om varje metod och dess parametrar, använd mssparkutils.lakehouse.help("methodName") funktionen.

Genom att använda MSSparkUtils Lakehouse-verktyg kan du mer effektivt hantera dina Lakehouse-produkter och integrera denna hantering i dina Fabric-pipelines, vilket förbättrar din totala datahanteringsupplevelse.

Utforska dessa verktyg och integrera dem i dina Fabric-arbetsflöden för smidig lakehouse-produkthantering.

Runtime-verktyg

Visa sessionskontextinformationen

Genom att använda mssparkutils.runtime.context, kan du få kontextinformation för den aktuella live-sessionen, inklusive anteckningsbokens namn, standard lakehouse, arbetsytsinformation, om det är en pipeline-körning och mer.

mssparkutils.runtime.context

Kända problem

När du använder en runtime-version som är senare än 1.2 och kör mssparkutils.help(), stöds inte de listade fabricClient-, warehouse- och workspace-API :erna för tillfället.

Relaterat innehåll

• Bibliotekshantering