Utilitas Microsoft Spark (MSSparkUtils) untuk Fabric

Microsoft Spark Utilities (MSSparkUtils) adalah paket bawaan untuk membantu Anda melakukan tugas umum dengan mudah. Anda dapat menggunakan MSSparkUtils untuk bekerja dengan sistem file, untuk mendapatkan variabel lingkungan, untuk mengikat notebook bersama-sama, dan bekerja dengan rahasia. Paket MSSparkUtils tersedia di PySpark (Python) Scala, notebook SparkR, dan alur Fabric.

Utilitas sistem file

mssparkutils.fs menyediakan utilitas untuk bekerja dengan berbagai sistem file, termasuk Azure Data Lake Storage (ADLS) Gen2 dan Azure Blob Storage. Pastikan Anda mengonfigurasi akses ke Azure Data Lake Storage Gen2 dan Azure Blob Storage dengan tepat.

Jalankan perintah berikut untuk mendapatkan gambaran umum metode yang tersedia:

from notebookutils import mssparkutils
mssparkutils.fs.help()

Hasil

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 bekerja dengan sistem file dengan cara yang sama seperti API Spark. Ambil penggunaan mssparkuitls.fs.mkdirs() dan Fabric lakehouse misalnya:

Daftar file

Untuk mencantumkan konten direktori, gunakan mssparkutils.fs.ls('Jalur direktori Anda'). Contohnya:

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 

Perlihatkan properti file

Metode ini mengembalikan properti file termasuk nama file, jalur file, ukuran file, dan apakah itu direktori dan file.

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

Buat direktori baru

Metode ini membuat direktori yang diberikan jika tidak ada, dan membuat direktori induk yang diperlukan.

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 

Salin file

Metode ini menyalin file atau direktori, dan mendukung aktivitas salin di seluruh sistem file.

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

File salin berkinerja

Metode ini menyediakan cara penyalinan atau pemindahan file yang lebih cepat, terutama data dalam volume besar.

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

Pratinjau konten file

Metode ini mengembalikan hingga byte 'maxBytes' pertama dari file yang diberikan sebagai String yang dikodekan dalam UTF-8.

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

Pindahkan file

Metode ini memindahkan file atau direktori, dan mendukung perpindahan di seluruh sistem file.

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.

Tulis file

Metode ini menulis string yang diberikan ke file, dikodekan dalam 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

Tambahkan konten ke file

Metode ini menambahkan string yang diberikan ke file, yang dikodekan dalam 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

Hapus file atau direktori

Metode ini menghapus file atau direktori.

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

Memasang/melepas direktori

Temukan informasi selengkapnya tentang penggunaan terperinci di Pemasangan file dan lepaskan.

Utilitas notebook

Gunakan Utilitas Notebook MSSparkUtils untuk menjalankan buku catatan atau keluar dari buku catatan dengan nilai. Jalankan perintah berikut untuk mendapatkan gambaran umum tentang metode yang tersedia:

mssparkutils.notebook.help()

Output:

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.

Referensikan notebook

Metode ini mereferensikan buku catatan dan mengembalikan nilai keluarnya. Anda bisa menjalankan panggilan fungsi nesting di notebook secara interaktif atau dalam alur. Buku catatan yang dirujuk berjalan pada kumpulan Spark buku catatan yang memanggil fungsi ini.

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

Contohnya:

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

Fabric notebook juga mendukung referensi notebook di beberapa ruang kerja dengan menentukan ID ruang kerja.

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

Anda dapat membuka tautan rekam jepret dari referensi yang dijalankan dalam output sel. Rekam jepret mengambil hasil eksekusi kode dan memungkinkan Anda untuk dengan mudah men-debug eksekusi referensi.

Referensi menjalankan beberapa buku catatan secara paralel

Metode mssparkutils.notebook.runMultiple() ini memungkinkan Anda menjalankan beberapa notebook secara paralel atau dengan struktur topologi yang telah ditentukan sebelumnya. API menggunakan mekanisme implementasi multi-utas dalam sesi spark, yang berarti sumber daya komputasi dibagikan oleh notebook referensi yang dijalankan.

Dengan mssparkutils.notebook.runMultiple(), Anda dapat:

• Jalankan beberapa notebook secara bersamaan, tanpa menunggu masing-masing selesai.

• Tentukan dependensi dan urutan eksekusi untuk notebook Anda, menggunakan format JSON sederhana.

• Optimalkan penggunaan sumber daya komputasi Spark dan kurangi biaya proyek Fabric Anda.

• Tampilkan Rekam jepret dari setiap catatan eksekusi buku catatan dalam output, dan debug/pantau tugas buku catatan Anda dengan mudah.

• Dapatkan nilai keluar dari setiap aktivitas eksekutif dan gunakan dalam tugas hilir.

Anda juga dapat mencoba menjalankan mssparkutils.notebook.help("runMultiple") untuk menemukan contoh dan penggunaan terperinci.

Berikut adalah contoh sederhana menjalankan daftar buku catatan secara paralel menggunakan metode ini:

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

Hasil eksekusi dari buku catatan akar adalah sebagai berikut:

Berikut ini adalah contoh menjalankan notebook dengan struktur topologis menggunakan mssparkutils.notebook.runMultiple(). Gunakan metode ini untuk mengatur notebook dengan mudah melalui pengalaman kode.

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

Hasil eksekusi dari buku catatan akar adalah sebagai berikut:

Keluar dari notebook

Metode ini keluar dari buku catatan dengan nilai. Anda bisa menjalankan panggilan fungsi nesting di notebook secara interaktif atau dalam alur.

• Saat Anda memanggil fungsi exit() dari notebook secara interaktif, notebook Fabric melempar pengecualian, melompati sel berikutnya, dan menjaga sesi Spark tetap hidup.

• Saat Anda mengatur buku catatan dalam alur yang memanggil fungsi exit(), aktivitas notebook kembali dengan nilai keluar, menyelesaikan eksekusi alur, dan menghentikan sesi Spark.

• Ketika Anda memanggil fungsi exit() di notebook yang sedang dirujuk, Fabric Spark akan menghentikan eksekusi lebih lanjut dari notebook yang dirujuk, dan terus menjalankan sel berikutnya di notebook utama yang memanggil fungsi run(). Misalnya: Notebook1 memiliki tiga sel dan memanggil fungsi exit() di sel kedua. Notebook2 memiliki lima sel dan panggilan run(notebook1) di sel ketiga. Saat Anda menjalankan Notebook2, Notebook1 berhenti di sel kedua saat menekan fungsi exit(). Notebook2 terus menjalankan sel keempat dan sel kelimanya.

mssparkutils.notebook.exit("value string")

Contohnya:

Buku catatan Sample1 dengan dua sel berikut:

• Sel 1 menentukan parameter input dengan nilai default yang diatur ke 10.

• Sel 2 keluar dari buku catatan dengan input sebagai nilai keluar.

Anda bisa menjalankan Sample1 di notebook lain dengan nilai default:

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

Output:

Notebook executed successfully with exit value 10

Anda bisa menjalankan Sample1 di notebook lainnya dan mengatur nilai input sebagai 20:

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

Output:

Notebook executed successfully with exit value 20

Utilitas kredensial

Anda dapat menggunakan MSSparkUtils Credentials Utilities untuk mendapatkan token akses dan mengelola rahasia di Azure Key Vault.

Jalankan perintah berikut untuk mendapatkan gambaran umum tentang metode yang tersedia:

mssparkutils.credentials.help()

Output:

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

Dapatkan token

getToken mengembalikan token Microsoft Entra untuk audiens dan nama tertentu (opsional). Daftar berikut ini memperlihatkan kunci audiens yang saat ini tersedia:

• Sumber Daya Audiens Penyimpanan: "penyimpanan"
• Sumber Daya Power BI: "pbi"
• Sumber Daya Azure Key Vault: "keyvault"
• Sumber Daya Synapse RTA KQL DB: "kusto"

Jalankan perintah berikut untuk mendapatkan token:

mssparkutils.credentials.getToken('audience Key')

Dapatkan rahasia dengan menggunakan kredensial pengguna

getSecret mengembalikan rahasia Azure Key Vault untuk titik akhir Azure Key Vault tertentu dan nama rahasia menggunakan kredensial pengguna.

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

Pemasangan dan pelampiran file

Fabric mendukung skenario pemasangan berikut dalam paket Utilitas Microsoft Spark. Anda dapat menggunakan API mount, unmount, getMountPath(), dan mounts() untuk melampirkan penyimpanan jarak jauh (ADLS Gen2) ke semua node yang berfungsi (node driver dan node pekerja). Setelah titik pemasangan penyimpanan ada, gunakan API file lokal untuk mengakses data seolah-olah disimpan dalam sistem file lokal.

Cara memasang akun ADLS Gen2

Contoh berikut mengilustrasikan cara memasang Azure Data Lake Storage Gen2. Pemasangan Blob Storage juga berfungsi serupa.

Contoh ini mengasumsikan bahwa Anda memiliki satu akun Data Lake Storage Gen2 bernama storegen2, dan akun memiliki satu kontainer bernama mycontainer yang ingin Anda pasang ke /uji ke sesi Spark notebook Anda.

Untuk memasang kontainer yang disebut mycontainer, mssparkutils terlebih dahulu perlu memeriksa apakah Anda memiliki izin untuk mengakses kontainer. Saat ini, Fabric mendukung dua metode autentikasi untuk operasi pemasangan pemicu: accountKey dan sastoken.

Memasang melalui token tanda tangan akses bersama atau kunci akun

MSSparkUtils mendukung secara eksplisit meneruskan kunci akun atau token Tanda tangan akses bersama (SAS) sebagai parameter untuk memasang target.

Untuk alasan keamanan, kami sarankan Anda menyimpan kunci akun atau token SAS di Azure Key Vault (seperti yang ditunjukkan cuplikan layar berikut). Anda kemudian dapat mengambilnya dengan menggunakan mssparkutils.credentials.getSecret API. Untuk informasi selengkapnya tentang Azure Key Vault, lihat Tentang kunci akun penyimpanan terkelola Azure Key Vault.

Kode sampel untuk metode 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}
)

Contoh kode untuk 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

Parameter pemasangan:

• fileCacheTimeout: Blob akan di-cache di folder sementara lokal selama 120 detik secara default. Selama waktu ini, blobfuse tidak akan memeriksa apakah file sudah diperbarui atau tidak. Parameter dapat diatur untuk mengubah waktu habis default. Ketika beberapa klien memodifikasi file secara bersamaan, untuk menghindari inkonsistensi antara file lokal dan jarak jauh, sebaiknya persingkat waktu cache, atau bahkan mengubahnya menjadi 0, dan selalu mendapatkan file terbaru dari server.
• batas waktu: Batas waktu operasi pemasangan adalah 120 detik secara default. Parameter dapat diatur untuk mengubah waktu habis default. Ketika ada terlalu banyak pelaksana atau ketika waktu pemasangan habis, sebaiknya tingkatkan nilainya.

Anda dapat menggunakan parameter ini seperti ini:

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

Cara memasang lakehouse

Kode sampel untuk memasang lakehouse ke /test:

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

Mengakses file di bawah titik pemasangan dengan menggunakan mssparktuils fs API

Tujuan utama operasi pemasangan adalah untuk memungkinkan pelanggan mengakses data yang disimpan di akun penyimpanan jarak jauh dengan API sistem file lokal. Anda juga dapat mengakses data dengan menggunakan mssparkutils fs API dengan jalur yang dipasang sebagai parameter. Format jalur ini sedikit berbeda.

Asumsikan bahwa Anda memasang mycontainer kontainer Data Lake Storage Gen2 ke /test dengan menggunakan API pemasangan. Saat Anda mengakses data dengan API sistem file lokal, format jalurnya seperti ini:

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

Saat Anda ingin mengakses data dengan menggunakan mssparkutils fs API, sebaiknya gunakan getMountPath() untuk mendapatkan jalur yang akurat:

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

• Membuat daftar direktori:

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

• Membaca konten file:

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

• Membuat direktori:

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

Mengakses file di bawah titik pemasangan melalui jalur lokal

Anda dapat dengan mudah membaca dan menulis file di titik pemasangan menggunakan sistem file standar. Berikut adalah contoh 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"))

Cara memeriksa titik pemasangan yang ada

Anda dapat menggunakan API mssparkutils.fs.mounts() untuk memeriksa semua info titik pemasangan yang ada:

mssparkutils.fs.mounts()

Cara melepaskan titik pemasangan

Gunakan kode berikut untuk melepas titik pemasangan Anda (/uji dalam contoh ini):

mssparkutils.fs.unmount("/test")

Pembatasan yang diketahui

• Pemasangan saat ini adalah konfigurasi tingkat pekerjaan; kami sarankan Anda menggunakan API pemasangan untuk memeriksa apakah ada titik pemasangan atau tidak tersedia.

• Mekanisme lepas tidak otomatis. Ketika aplikasi berjalan selesai, untuk melepas titik pemasangan dan melepaskan ruang disk, Anda perlu secara eksplisit memanggil API lepaskan dalam kode Anda. Jika tidak, titik pemasangan masih akan ada di node setelah eksekusi aplikasi selesai.

• Memasang akun penyimpanan ADLS Gen1 tidak didukung.

Utilitas Lakehouse

mssparkutils.lakehouse menyediakan utilitas yang khusus disesuaikan untuk mengelola artefak Lakehouse. Utilitas ini memberdayakan pengguna untuk membuat, mengambil, memperbarui, dan menghapus artefak Lakehouse dengan mudah.

Gambaran umum metode

Di bawah ini adalah gambaran umum metode yang tersedia yang disediakan oleh 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]

Contoh penggunaan

Untuk menggunakan metode ini secara efektif, pertimbangkan contoh penggunaan berikut:

Membuat artefak Lakehouse

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

Mengambil Artefak Lakehouse

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

Memperbarui artefak Lakehouse

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

Menghapus artefak Lakehouse

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

Mencantumkan artefak Lakehouse

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

Informasi Tambahan

Untuk informasi lebih rinci tentang setiap metode dan parameternya, gunakan mssparkutils.lakehouse.help("methodName") fungsi .

Dengan utilitas Lakehouse MSSparkUtils, mengelola artefak Lakehouse Anda menjadi lebih efisien dan terintegrasi ke dalam alur Fabric Anda, meningkatkan pengalaman manajemen data Anda secara keseluruhan.

Jangan ragu untuk menjelajahi utilitas ini dan memasukkannya ke dalam alur kerja Fabric Anda untuk manajemen artefak Lakehouse yang mulus.

Utilitas runtime

Perlihatkan info konteks sesi

Dengan mssparkutils.runtime.context Anda bisa mendapatkan informasi konteks sesi langsung saat ini, termasuk nama notebook, lakehouse default, info ruang kerja, jika itu adalah eksekusi alur, dll.

mssparkutils.runtime.context

Masalah yang diketahui

Saat menggunakan versi runtime di atas 1.2 dan menjalankan mssparkutils.help(), FABRICClient, gudang, dan API ruang kerja yang tercantum tidak didukung untuk saat ini, akan tersedia di lebih lanjut.

Konten terkait

• Manajemen pustaka