Microsoft Spark Utilities (MSSparkUtils) voor Fabric

Microsoft Spark Utilities (MSSparkUtils) is een ingebouwd pakket dat je helpt om veelvoorkomende taken eenvoudig uit te voeren. Gebruik MSSparkUtils om te werken met bestandssystemen, omgevingsvariabelen op te halen, notebooks te koppelen en met geheimen te werken. Het MSSparkUtils-pakket is beschikbaar in PySpark (Python), Scala, SparkR notebooks en Fabric pipelines.

Hulpprogramma's voor bestandssysteem

mssparkutils.fs biedt hulpprogramma's voor het werken met verschillende bestandssystemen, waaronder Azure Data Lake Storage Gen2 en Azure Blob Storage. Zorg ervoor dat u de toegang tot Azure Data Lake Storage Gen2 en Azure Blob Storage op de juiste manier configureert.

Voer de volgende opdrachten uit voor een overzicht van de beschikbare methoden:

from notebookutils import mssparkutils
mssparkutils.fs.help()

Uitvoer

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 werkt op dezelfde manier met het bestandssysteem als Spark-API's. Neem bijvoorbeeld mssparkuitls.fs.mkdirs() en lakehouse-gebruik:

Bestanden in een lijst weergeven

Als u de inhoud van een map wilt weergeven, gebruikt u mssparkutils.fs.ls('Uw mappad'). Voorbeeld:

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 

Bestandseigenschappen weergeven

Deze methode geeft bestandseigenschappen terug, waaronder de bestandsnaam, het pad, de bestandsgrootte en of het een map of een bestand is.

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

Nieuwe map maken

Deze methode maakt de gespecificeerde map aan als deze niet bestaat, en maakt eventuele benodigde oudermappen.

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 

Bestand kopiëren

Met deze methode wordt een bestand of map gekopieerd en worden kopieeractiviteiten in bestandssystemen ondersteund.

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

Efficiënt bestand kopiëren

Deze methode biedt een snellere manier om bestanden te kopiëren of te verplaatsen, met name grote hoeveelheden gegevens.

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

Voorbeeld van bestandsinhoud

Deze methode levert tot de eerste maxBytes bytes van het opgegeven bestand terug als een string gecodeerd in UTF-8.

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

Bestand verplaatsen

Met deze methode wordt een bestand of map verplaatst en worden verplaatsingen tussen bestandssystemen ondersteund.

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.

Bestand schrijven

Met deze methode wordt de opgegeven tekenreeks weggeschreven naar een bestand, gecodeerd in 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

Inhoud toevoegen aan een bestand

Met deze methode wordt de opgegeven tekenreeks toegevoegd aan een bestand, gecodeerd in 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

Bestand of map verwijderen

Met deze methode wordt een bestand of map verwijderd.

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

Map koppelen/ontkoppelen

Voor meer informatie over gedetailleerd gebruik, zie File mount and unmount.

Hulpprogramma's voor notebooks

Gebruik de MSSparkUtils-notebookhulpprogramma's om een notebook uit te voeren of een notebook met een waarde af te sluiten. Voer de volgende opdracht uit om een overzicht te krijgen van de beschikbare methoden:

mssparkutils.notebook.help()

Uitvoer:

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.

Verwijzen naar een notitieblok

Deze methode verwijst naar een notebook en retourneert de uitvoerwaarde. U kunt geneste functie-aanroepen in een notebook interactief of in een pijplijn uitvoeren. Het notebook waarnaar wordt verwezen draait op de Spark-pool van het notebook dat deze functie aanroept.

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

Voorbeeld:

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

Fabric Notebook ondersteunt ook verwijzingen naar notebooks in meerdere werkruimten door de werkruimte-id op te geven.

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

U kunt de koppeling naar de momentopname van de referentierun openen binnen de celuitvoer. Met de momentopname worden de resultaten van de codeuitvoering vastgelegd en kunt u eenvoudig fouten opsporen in een verwijzingsuitvoering.

Naslaginformatie over het parallel uitvoeren van meerdere notebooks

Met de methode mssparkutils.notebook.runMultiple() kunt u meerdere notebooks parallel of met een vooraf gedefinieerde topologische structuur uitvoeren. De API gebruikt een multithreaded implementatie om kindnotitieboeken in te dienen, in de wachtrij te zetten en te monitoren die worden uitgevoerd op geïsoleerde REPL-instanties (read-eval-print-loop) binnen de bestaande Spark-sessie. De verwezen kindnotitieboeken delen de rekenkrachten van de sessie.

Met mssparkutils.notebook.runMultiple()kunt u het volgende doen:

• Voer meerdere notebooks tegelijk uit, zonder te wachten tot elke notebook is voltooid.

• Geef de afhankelijkheden en de volgorde van uitvoering voor uw notebooks op met behulp van een eenvoudige JSON-indeling.

• Optimaliseer het gebruik van Spark-rekenresources en verlaag de kosten van uw Fabric-projecten.

• Bekijk de snapshots van elk notebook-run record in de output, en debug en monitor je notebooktaken op een handige manier.

• Haal de afsluitwaarde van elke uitvoerende activiteit op en gebruik deze in downstream taken.

U kunt ook proberen de mssparkutils.notebook.help("runMultiple") uit te voeren om het voorbeeld en het gedetailleerde gebruik te vinden.

Hier volgt een eenvoudig voorbeeld van het parallel uitvoeren van een lijst met notebooks met behulp van deze methode:

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

Het uitvoerresultaat van het hoofdnotebook is als volgt:

Het volgende voorbeeld laat zien dat notitieboeken met een topologische structuur worden uitgevoerd door gebruik te maken van mssparkutils.notebook.runMultiple(). Gebruik deze methode om eenvoudig notebooks te organiseren via een code-ervaring.

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

Het uitvoerresultaat van het hoofdnotebook is als volgt:

Een notitieblok afsluiten

Met deze methode verlaat je een notebook met een waarde. U kunt geneste functie-aanroepen in een notebook interactief of in een pijplijn uitvoeren.

• Wanneer u een exit()-functie vanuit een notebook interactief aanroept, gooit het Fabric-notebook een uitzondering, slaat het het uitvoeren van de volgende cellen over en blijft de Spark-sessie actief.

• Wanneer u een notebook indeelt in een pijplijn die een exit() -functie aanroept, wordt de notebookactiviteit geretourneerd met een afsluitwaarde, wordt de pijplijnuitvoering voltooid en wordt de Spark-sessie gestopt. Sluit de exit() -functie niet rond een try/catch, want deze NotebookExit-uitzondering moet zich voortplanten zodat de pijplijn de returnwaarde kan krijgen.

• Wanneer je een exit()-functie aanroept in een notebook dat wordt gerefereerd, stopt Fabric Spark de verdere uitvoering van het referentiede notebook en blijft het de volgende cellen in het hoofdnotebook uitvoeren die de run()-functie aanroepen. Bijvoorbeeld: Notebook1 heeft drie cellen en roept een exit() -functie aan in de tweede cel. Notebook2 heeft vijf cellen en roept run(notebook1) aan in de derde cel. Wanneer u Notebook2 uitvoert, stopt Notebook1 in de tweede cel wanneer u de exit() -functie bereikt. Notebook2 blijft zijn vierde en vijfde cellen uitvoeren.

mssparkutils.notebook.exit("value string")

Voorbeeld:

Voorbeeld1 notebook met de volgende twee cellen:

• Cel 1 definieert een invoerparameter met de standaardwaarde ingesteld op 10.

• Cel 2 verlaat het notitieblok met invoer als uitvoerwaarde.

U kunt het voorbeeld1 uitvoeren in een ander notebook met standaardwaarden:

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

Uitvoer:

Notebook executed successfully with exit value 10

U kunt het voorbeeld1 uitvoeren in een ander notebook en de invoerwaarde instellen als 20:

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

Uitvoer:

Notebook executed successfully with exit value 20

Hulpprogramma's voor inloggegevens

Je kunt de MSSparkUtils Credentials Utilities gebruiken om toegangstokens te krijgen en geheimen te beheren in Azure Key Vault.

Voer de volgende opdracht uit om een overzicht te krijgen van de beschikbare methoden:

mssparkutils.credentials.help()

Uitvoer:

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

Token ophalen

getTokengeeft een Microsoft Entra-token terug voor een bepaalde doelgroep en naam (optioneel). In de volgende lijst ziet u de momenteel beschikbare doelgroepsleutels:

• Resource voor opslagdoelgroep: storage
• Power BI-resource: pbi
• Azure Key Vault Resource:keyvault
• Synapse RTA KQL DB Resource: kusto

Voer de volgende opdracht uit om het token op te halen:

mssparkutils.credentials.getToken('audience Key')

Secret ophalen met gebruikersreferenties

getSecretgeeft een Azure Key Vault-geheim terug voor een bepaald Azure Key Vault-endpoint en geheime naam door gebruik te maken van gebruikersgegevens.

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

Bestandskoppeling en ontkoppelen

Fabric ondersteunt de volgende koppelscenario's in het Microsoft Spark Utilities-pakket. Je kunt de mount, unmount,getMountPath() en mounts() API's gebruiken om externe opslag (Azure Data Lake Storage Gen2) aan alle werkende nodes (driver node en worker nodes) te koppelen. Nadat het opslagkoppelingspunt is ingesteld, gebruikt u de API voor het lokale bestand om toegang te krijgen tot gegevens alsof deze zijn opgeslagen in het lokale bestandssysteem.

Hoe je een Azure Data Lake Storage Gen2-account koppelt

Het volgende voorbeeld laat zien hoe je Azure Data Lake Storage Gen2 kunt mounten. Het koppelen van Blob Storage werkt op dezelfde manier.

In dit voorbeeld wordt ervan uitgegaan dat u één Data Lake Storage Gen2-account met de naam storegen2 hebt en dat het account één container heeft met de naam mycontainer die u wilt koppelen aan /test in uw Spark-notebooksessie.

Om de container genaamd mycontainer te mounten, controleert mssparkutils eerst of je toestemming hebt om toegang te krijgen tot de container. Fabric ondersteunt drie authenticatiemethoden voor de triggermount-operatie: Microsoft Entra token (standaard en aanbevolen), accountKey en sastoken. Raadpleeg notebookutils voor meer informatie over Microsoft Entra tokenauthenticatie en de huidige API.

Mount door gebruik te maken van een shared access-handtekeningtoken of accountsleutel

MSSparkUtils ondersteunt het expliciet doorgeven van een accountsleutel of SAS-token (Shared Access Signature) als parameter om de doellocatie te koppelen.

Om veiligheidsredenen raden we u aan accountsleutels of SAS-tokens op te slaan in Azure Key Vault (zoals in de volgende schermopname wordt weergegeven). U kunt ze vervolgens ophalen met behulp van de mssparkutils.credentials.getSecret-API . Voor meer informatie over Azure Key Vault, zie Over door Azure Key Vault beheerde opslagaccountsleutels.

Voorbeeldcode voor de accountKey-methode :

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

Voorbeeldcode voor 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

Parameters koppelen:

• fileCacheTimeout: Blobs worden standaard 120 seconden in de lokale tijdelijke map in de cache opgeslagen. Gedurende deze tijd controleert blobfuse niet of het bestand up-to-date is. Stel deze parameter in om de standaard time-out te wijzigen. Wanneer meerdere clients tegelijkertijd bestanden wijzigen, raden we aan om inconsistenties tussen lokale en externe bestanden te voorkomen, de cachetijd te verkorten of zelfs naar 0 te zetten en altijd de nieuwste bestanden van de server te halen.
• timeout: De time-out voor mount-operaties is standaard 120 seconden. Stel deze parameter in om de standaard time-out te wijzigen. Wanneer er te veel executeurs zijn of wanneer de mount uitloopt, raden we aan de waarde te verhogen.

U kunt deze parameters als volgt gebruiken:

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

Hoe een lakehouse te monteren

Voorbeeldcode voor het monteren van een meerhuis op /test:

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

Open bestanden onder het koppelpunt met behulp van de mssparkutils fs-API

Het hoofddoel van de mount-operatie is om toegang te krijgen tot de gegevens die in een externe opslagrekening zijn opgeslagen via een lokale bestandssysteem-API. U kunt de gegevens ook openen met behulp van de mssparkutils fs-API met een gekoppeld pad als parameter. Deze padindeling is iets anders.

Stel dat je de Data Lake Storage Gen2-container mycontainer hebt gekoppeld aan /test met behulp van de mount-API. Wanneer je de data benadert via een lokale bestandssysteem-API, is het padformaat als volgt:

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

Wanneer je toegang wilt krijgen tot de data via de mssparkutils fs API, raden we aan getMountPath() te gebruiken om het nauwkeurige pad te krijgen:

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

• Directorieën weergeven:

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

• Bestandsinhoud lezen:

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

• Maak een map aan:

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

Toegang tot bestanden onder het mount point via het lokale pad

U kunt de bestanden eenvoudig lezen en schrijven op het aankoppelpunt met behulp van het standaard bestandssysteem. Hier volgt een Python-voorbeeld:

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

Bestaande koppelpunten controleren

U kunt mssparkutils.fs.mounts() API gebruiken om alle bestaande koppelingspuntgegevens te controleren:

mssparkutils.fs.mounts()

Hoe je het koppelpunt kunt ontkoppelen

Gebruik de volgende code om het koppelpunt los te koppelen (/test in dit voorbeeld):

mssparkutils.fs.unmount("/test")

Bekende beperkingen

• De huidige koppeling is een configuratie op taakniveau. We raden aan om de mounts API te gebruiken om te controleren of er een mountpoint bestaat of niet.

• Het ontkoppelingsmechanisme is niet automatisch. Wanneer de uitvoering van de toepassing is voltooid, moet u expliciet een ontkoppel-API in uw code oproepen om de koppeling op te heffen en de schijfruimte vrij te maken. Anders bestaat het koppelpunt nog steeds in het knooppunt nadat de uitvoering van de toepassing is voltooid.

• Het aanmaken van een Azure Data Lake Storage Gen1 opslagaccount wordt niet ondersteund.

Lakehouse-hulpprogramma's

De mssparkutils.lakehouse module biedt nutsvoorzieningen voor het beheren van lakehouse-items. Deze hulpprogramma's maken het eenvoudig om lakehouse-items aan te maken, op te halen, bij te werken en te verwijderen.

Overzicht van methoden

De volgende methoden zijn beschikbaar in de mssparkutils.lakehouse module:

# 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]

Voorbeelden van gebruik

Om deze methoden effectief te gebruiken, overweeg de volgende gebruiksvoorbeelden:

Een meerhuis-item maken

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

Een item in een meerhuis terughalen

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

Een item bijwerken in een lakehouse

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

Een item verwijderen in een lakehouse

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

Vermelding van meerhuis-items

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

Aanvullende informatie

Voor meer gedetailleerde informatie over elke methode en haar parameters, gebruik de mssparkutils.lakehouse.help("methodName") functie.

Door gebruik te maken van de Lakehouse-nutsvoorzieningen van MSSparkUtils, kunt u uw lakehouse-items efficiënter beheren en dit beheer integreren in uw Fabric-pijplijnen, wat uw algehele datamanagementervaring verbetert.

Ontdek deze hulpmiddelen en verwerk ze in je Fabric-workflows voor naadloos lakehouse-itembeheer.

Runtimehulpprogramma's

De sessiecontextgegevens weergeven

Door mssparkutils.runtime.context te gebruiken, kun je contextinformatie ophalen over de huidige livesessie, waaronder de naam van het notebook, het standaardlakehouse, werkruimtegegevens, of het om een pijplijnuitvoering gaat, en meer.

mssparkutils.runtime.context

Bekend probleem

Wanneer je een runtime-versie gebruikt die later is dan 1.2 en draaitmssparkutils.help(), worden de vermelde fabricClient-, warehouse- en workspace-API's momenteel niet ondersteund.

Gerelateerde inhoud

• Bibliotheekbeheer