Microsoft Spark Utilities (MSSparkUtils) pre Fabric

Microsoft Spark Utilities (MSSparkUtils) je vstavaný balík, ktorý vám pomôže jednoducho vykonávať bežné úlohy. Funkciu MSSparkUtils môžete používať na prácu so súborovými systémami, získavanie premenných prostredia, reťazové poznámkové bloky a prácu s tajomstvámi. Balík MSSparkUtils je k dispozícii v kanáloch PySpark (Python) Scala, Poznámkové bloky SparkR a Fabric.

Pomôcky systému súborov

mssparkutils.fs poskytuje pomôcky na prácu s rôznymi súborovými systémami vrátane služby Azure Data Lake Storage (ADLS) Gen2 a služby Azure Blob Storage. Nezabudnite nakonfigurovať prístup k službe Azure Data Lake Storage Gen2 a úložisku Azure Blob Storage správne.

Spustite nasledujúce príkazy a získajte prehľad dostupných metód:

from notebookutils import mssparkutils
mssparkutils.fs.help()

Výkon

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.

Služba MSSparkUtils funguje so systémom súborov rovnakým spôsobom ako rozhrania Spark API. Vezmite mssparkuitls.fs.mkdirs () a využitie jazera Fabric, napríklad:

Súbory zoznamu

Ak chcete zobraziť zoznam obsahu adresára, použite mssparkutils.fs.ls('Cesta k adresáru'). Napríklad:

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 

Zobrazenie vlastností súboru

Táto metóda vráti vlastnosti súboru vrátane názvu súboru, cesty k súboru, veľkosti súboru a toho, či ide o adresár a súbor.

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

Vytvorenie nového adresára

Táto metóda vytvorí daný adresár, ak neexistuje, a vytvorí všetky potrebné nadradené adresáre.

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 

Kopírovať súbor

Táto metóda skopíruje súbor alebo adresár a podporuje aktivitu kopírovania v súborových systémoch.

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 copy file

Táto metóda poskytuje rýchlejší spôsob kopírovania alebo premiestňovania súborov, najmä veľkých objemov údajov.

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

Zobraziť ukážku obsahu súboru

Táto metóda vráti až prvé bajty "maxBytes" daného súboru ako reťazec kódovaný v UTF-8.

mssparkutils.fs.head('file path', maxBytes to read)

Presunutie súboru

Táto metóda premiestni súbor alebo adresár a podporuje presuny do súborových systémov.

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.

Zapisovať súbor

Táto metóda zapíše daný reťazec do súboru, ktorý je kódovaný v 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

Pripojenie obsahu k súboru

Táto metóda pripojí daný reťazec k súboru, ktorý je kódovaný v 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

Odstránenie súboru alebo adresára

Táto metóda odstráni súbor alebo adresár.

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

Pripojenie alebo zrušenie pripojenia adresára

Ďalšie informácie o podrobnom používaní nájdete v téme Pripojenie a zrušenie pripojenia súboru.

Pomôcky pre poznámkové bloky

Pomocou pomôcok poznámkového bloku služby MSSparkUtils môžete spustiť poznámkový blok alebo ukončiť poznámkový blok s hodnotou. Spustením nasledujúceho príkazu získate prehľad dostupných metód:

mssparkutils.notebook.help()

Výstup:

exit(value: String): void -> 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.

Odkaz na poznámkový blok

Táto metóda odkazuje na poznámkový blok a vráti jeho výstupnú hodnotu. Volania vnorených funkcií môžete spustiť interaktívne v poznámkovom bloke alebo v kanáli. Poznámkový blok, na ktorý sa odkazuje, je spustený vo fonde spark poznámkového bloku, ktorý túto funkciu volá.

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

Napríklad:

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

Poznámkový blok služby Fabric tiež podporuje odkazovanie na poznámkové bloky vo viacerých pracovných priestoroch zadaním ID pracovného priestoru.

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

Prepojenie na snímku odkazu môžete otvoriť v bunke výstupe. Snímka zaznamenáva výsledky spustenia kódu a umožňuje jednoducho ladiť referenčné spustenie.

Odkaz na paralelné spustenie viacerých poznámkových blokov

Metóda mssparkutils.notebook.runMultiple() vám umožňuje spustiť viaceré poznámkové bloky paralelne alebo s preddefinovanou topologickou štruktúrou. Rozhranie API používa mechanizmus implementácie viacerých vlákien v rámci relácie spark, čo znamená, že výpočtové zdroje sú zdieľané spúšťačmi referenčných poznámkových blokov.

S možnosťou mssparkutils.notebook.runMultiple()môžete:

• Vykonajte viacero poznámkových blokov súčasne bez čakania na dokončenie každého z nich.

• Zadajte závislosti a poradie vykonávania pre poznámkové bloky pomocou jednoduchého formátu JSON.

• Optimalizujte použitie výpočtových zdrojov služby Spark a znížte náklady na projekty služby Fabric.

• Zobrazte snímky každého záznamu spúšťania poznámkového bloku vo výstupe a pohodlné ladenie/monitorovanie úloh poznámkového bloku.

• Získajte hodnotu výstupu každej aktivity vedúceho pracovníka a použite ich v následných úlohách.

Môžete tiež skúsiť spustiť mssparkutils.notebook.help("runMultiple") a nájsť príklad a podrobné použitie.

Tu je jednoduchý príklad paralelného spustenia zoznamu poznámkových blokov pomocou tejto metódy:

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

Výsledok spustenia koreňového poznámkového bloku je nasledovný:

V nasledujúcom príklade je uvedený príklad spustenia poznámkových blokov s topologickou štruktúrou pomocou .mssparkutils.notebook.runMultiple() Pomocou tejto metódy môžete jednoducho koordinovať poznámkové bloky prostredníctvom prostredia kódu.

# 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, default to 50
}
mssparkutils.notebook.runMultiple(DAG, {"displayDAGViaGraphviz": False})

Výsledok spustenia koreňového poznámkového bloku je nasledovný:

Ukončenie poznámkového bloku

Táto metóda ukončí poznámkový blok s hodnotou. Volania vnorených funkcií môžete spustiť interaktívne v poznámkovom bloke alebo v kanáli.

• Pri interaktívnom volaní funkcie exit() z notebooku poznámkový blok služby Fabric vyvolá výnimku, vynechá spúšťanie nasledujúcich buniek a udržiava reláciu Spark nažive.

• Pri koordinovaní poznámkového bloku v kanáli, ktorý volá funkciu exit( ), aktivita poznámkového bloku vráti hodnotu výstupu, dokončí spustenie kanála a zastaví reláciu služby Spark.

• Keď v notebooku , na ktorý sa odkazuje, vyvolá funkcia exit(), služba Fabric Spark zastaví ďalšie vykonávanie odkazovaného poznámkového bloku a bude naďalej spúšťať ďalšie bunky v hlavnom notebooku, ktorý volá funkciu run( ). Príklad: Notebook1 má tri bunky a v druhej bunke vyvolá funkciu exit( ). Notebook2 má päť buniek a v tretej bunke zavolá run(notebook1 ). Keď spustíte Notebook2, Notebook1 sa pri stlačení funkcie exit() zastaví na druhej bunke. Notebook2 naďalej spúšťa svoju štvrtú bunku a piatu bunku.

mssparkutils.notebook.exit("value string")

Napríklad:

Poznámkový blok Ukážka1 s nasledujúcimi dvomi bunkami:

• Bunka 1 definuje vstupný parameter s predvolenou hodnotou nastavenou na hodnotu 10.

• Bunka 2 ukončí poznámkový blok so vstupom ako výstupnou hodnotou.

Vzorku Sample1 môžete spustiť v inom poznámkovom bloku s predvolenými hodnotami:

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

Výstup:

Notebook executed successfully with exit value 10

Vzorku1 môžete spustiť v inom poznámkovom bloku a nastaviť vstupnú hodnotu ako 20:

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

Výstup:

Notebook executed successfully with exit value 20

Pomôcky pre prihlasovacie údaje

Pomocou pomôcok MSSparkUtils Credentials môžete získať prístupové tokeny a spravovať tajné kódy v službe Azure Key Vault.

Spustením nasledujúceho príkazu získate prehľad dostupných metód:

mssparkutils.credentials.help()

Výstup:

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

Získať token

Funkcia getToken vráti token Microsoft Entra pre danú cieľovú skupinu a meno (voliteľné). Nasledujúci zoznam zobrazuje aktuálne dostupné kľúče cieľovej skupiny:

• Storage Audience Resource: "storage"
• Zdroj služby Power BI: pbi
• Azure Key Vault Resource: "keyvault"
• Synapse RTA KQL DB Zdroj: "kusto"

Spustite nasledujúci príkaz, aby ste získali token:

mssparkutils.credentials.getToken('audience Key')

Získanie tajného kódu pomocou poverení používateľa

Funkcia getSecret vráti tajný kód služby Azure Key Vault pre daný koncový bod a meno tajného kódu služby Azure Key Vault pomocou prihlasovacích údajov používateľa.

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

Pripojenie a zrušenie pripojenia k súboru

Služba Fabric podporuje nasledujúce možnosti pripojenia v balíku služby Microsoft Spark Utilities. Na pripojenie vzdialeného úložiska (ADLS Gen2) k všetkým pracovným uzlom (uzly vodiča a uzly pracovníkov) môžete použiť rozhrania API pripojiť, zrušiť ich odnímanie, getMountPath() a pripojiť() rozhrania API. Po uložení bodu pripojenia ukladacieho priestoru použite rozhranie API lokálneho súboru na prístup k údajom, akoby boli uložené v lokálnom systéme súborov.

Ako pripojiť konto služby ADLS Gen2

V nasledujúcom príklade je znázornené, ako pripojiť Azure Data Lake Storage Gen2. Montážny úložný priestor objektu BLOB funguje podobne.

Tento príklad predpokladá, že máte jedno konto služby Data Lake Storage Gen2 s názvom storegen2 a konto má jeden kontajner s názvom mycontainer , ktorý chcete pripojiť k relácii Spark v notebooku /testovať .

Ak chcete pripojiť kontajner s názvom mycontainer, mssparkutils musí najskôr skontrolovať, či máte povolenie na prístup ku kontajneru. V súčasnosti služba Fabric podporuje dve metódy overovania pre operáciu pripojenia spúšťača: accountKey a sastoken.

Pripojenie prostredníctvom tokenu podpisu zdieľaného prístupu alebo kľúča konta

Služba MSSparkUtils podporuje explicitné odovzdanie kľúča konta alebo tokenu SAS (Shared access signature) ako parametra na pripojenie cieľa.

Z bezpečnostných dôvodov odporúčame uložiť kľúče kont alebo tokeny SAS v službe Azure Key Vault (ako je znázornené na nasledujúcej snímke obrazovky). Potom ich môžete načítať pomocou rozhrania mssparkutils.credentials.getSecret API. Ďalšie informácie o službe Azure Key Vault nájdete v téme Informácie o kľúčoch konta spravovaného úložiska Azure Key Vault.

Vzorový kód pre metódu 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}
)

Vzorový kód pre 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

Parametre pripojenia:

• fileCacheTimeout: Objekty BLOB sa predvolene ukladajú do vyrovnávacej pamäte v miestnom dočasnom priečinku na 120 sekúnd. Počas tohto obdobia objekt BLOBfuse nebude kontrolovať, či je súbor aktualizovaný alebo nie. Parameter možno nastaviť tak, aby sa zmenil predvolený čas časového limitu. Ak viacerí klienti upravujú súbory v rovnakom čase, odporúčame skrátiť čas vyrovnávacej pamäte alebo dokonca zmeniť ho na 0 a vždy získať najnovšie súbory zo servera, aby sa predišlo nekonzistenciám medzi lokálnymi a vzdialenými súbormi.
• časový limit: Časový limit operácie pripojenia je predvolene 120 sekúnd. Parameter možno nastaviť tak, aby sa zmenil predvolený čas časového limitu. Keď je príliš veľa spustiteľných vykonaní, alebo keď unikne čas pripojenia, odporúča sa zvýšiť hodnotu.

Nasledujúce parametre môžete použiť:

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

Ako pripojiť lakehouse

Vzorový kód pre montáž jazera na /test:

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

Prístup k súborom pod bodom pripojenia pomocou rozhrania mssparktuils fs API

Hlavným účelom operácie pripojenia je umožniť zákazníkom prístup k údajom uloženým vo vzdialenom konte úložiska s lokálnym rozhraním API systému súborov. K údajom môžete získať prístup aj pomocou rozhrania mssparkutils fs API s pripevnenou cestou ako parameter. Tento formát cesty je trochu iný.

Predpokladajme, že ste pomocou rozhrania API na pripojenie pripojili kontajner Data Lake Storage Gen2 ku kontajneru /test. Keď pristupujete k údajom pomocou rozhrania API lokálneho systému súborov, formát cesty je takýto:

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

Ak chcete získať prístup k údajom pomocou rozhrania mssparkutils fs API, na získanie presnej cesty odporúčame použiť funkciu getMountPath():

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

• Zoznam adresárov:

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

• Čítať obsah súboru:

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

• Vytvorte adresár:

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

Prístup k súborom pod bodom pripojenia cez lokálnu cestu

Súbory môžete jednoducho čítať a zapisovať do bodu pripojenia pomocou štandardného systému súborov. Tu je príklad jazyka Python:

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

Kontrola existujúcich bodov pripojenia

Rozhranie api mssparkutils.fs.mounts() môžete použiť na kontrolu všetkých existujúcich informácií o bode pripojenia:

mssparkutils.fs.mounts()

Ako zrušiť pripojenie bodu pripojenia

Na zrušenie pripojenia použite nasledujúci kód (/test v tomto príklade):

mssparkutils.fs.unmount("/test")

Známe obmedzenia

• Aktuálna konfigurácia na úrovni úlohy; Odporúčame použiť rozhranie API konektorov na kontrolu, či bod pripojenia existuje alebo nie je k dispozícii.

• Mechanizmus zrušenia presunutia nie je automatický. Po dokončení spustenia aplikácie musíte explicitne zavolať rozhranie API na zrušenie pripojenia a uvoľniť miesto na disku. V opačnom prípade bude bod pripojenia po dokončení spustenia aplikácie naďalej existovať v uzli.

• Montáž konta úložiska ADLS Gen1 nie je podporovaná.

Služby Lakehouse

mssparkutils.lakehouse Poskytuje pomôcky špeciálne prispôsobené na spravovanie artefaktov lakehouse. Tieto pomôcky umožňujú používateľom vytvárať, načítavať, aktualizovať a odstraňovať artefakty služby Lakehouse bez námahy.

Prehľad metód

Nižšie je uvedený prehľad dostupných metód poskytovaných pomocou mssparkutils.lakehousemetódy :

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

Príklady používania

Ak chcete efektívne využiť tieto metódy, zvážte nasledujúce príklady používania:

Vytvorenie artefaktu Lakehouse

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

Získanie artefaktu Lakehouse

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

Aktualizácia artefaktu Lakehouse

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

Odstránenie artefaktu Lakehouse

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

Výpis Lakehouse artefakty

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

Ďalšie informácie

Podrobné informácie o každej metóde a jej parametroch získate v téme o funkcii mssparkutils.lakehouse.help("methodName") .

Vďaka pomôckam MSSparkUtils lakehouse sa spravovanie artefaktov v službe Lakehouse stáva efektívnejším a integrovaným do vašich kanálov služby Fabric, čím sa zlepší možnosti celkovej správy údajov.

Neváhajte preskúmať tieto pomôcky a začleniť ich do svojich pracovných postupov služby Fabric pre bezproblémovú správu artefaktov Lakehouse.

Pomôcky modulu runtime

Zobrazenie kontextovej informácie relácie

Vďaka mssparkutils.runtime.context službe môžete získať kontextové informácie o aktuálnej živej relácii vrátane názvu poznámkového bloku, predvoleného prostredia jazera, informácií o pracovnom priestore, spustenia kanála atď.

mssparkutils.runtime.context

Známy problém

Pri použití verzie režimu runtime nad 1.2 a spustením mssparkutils.help()nie sú už uvedené rozhrania API pre tkaniny, sklad a pracovný priestor podporované, budú k dispozícii ďalej.

Súvisiaci obsah

• Správa knižníc