Doku için Microsoft Spark Yardımcı Programları (MSSparkUtils)

Microsoft Spark Utilities (MSSparkUtils), yaygın görevleri kolayca yerine getirmenize yardımcı olan yerleşik bir pakettir. Dosya sistemleriyle çalışmak, ortam değişkenlerini almak, not defterlerini birbirine zincirleme ve gizli dizilerle çalışmak için MSSparkUtils'i kullanın. MSSparkUtils paketi, PySpark (Python), Scala ve SparkR not defterlerinde ve Fabric işlem hatlarında kullanılabilir.

Not

• MsSparkUtils resmi olarak NotebookUtils olarak yeniden adlandırıldı. Mevcut kod geriye dönük uyumlu olacak ve hiçbir kırılma değişikliğine neden olmayacak. Sürekli destek ve yeni özelliklere erişim sağlamak için notebookutils'e yükseltmenizi şiddetle tavsiye ediyoruz. mssparkutils ad alanı gelecekte kullanımdan kaldırılacaktır.
• NotebookUtils, Spark 3.4 (Runtime v1.2) ve üzeri ile çalışacak şekilde tasarlanmıştır. Tüm yeni özellikler ve güncelleştirmeler, ileride notebookutils ad alanıyla özel olarak desteklenir.

Dosya sistemi yardımcı programları

mssparkutils.fs, Azure Data Lake Storage 2. Nesil ve Azure Blob Depolama dahil olmak üzere çeşitli dosya sistemleriyle çalışmak için yardımcı programlar sağlar. Azure Data Lake Storage 2. Nesil ve Azure Blob Depolama erişimini uygun şekilde yapılandırdığınızdan emin olun.

Kullanılabilir yöntemlere genel bakış için aşağıdaki komutları çalıştırın:

from notebookutils import mssparkutils
mssparkutils.fs.help()

Çıktı

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, dosya sistemiyle Spark API'leriyle aynı şekilde çalışır. Örneğin mssparkuitls.fs.mkdirs() ve göl evi kullanımını ele alalım:

Kullanım	HDFS kök dizininden göreli yol	ABFS dosya sistemi için mutlak yol	Sürücü düğümündeki yerel dosya sistemi için mutlak yol
Varsayılan Olmayan Lakehouse	Desteklenmez	mssparkutils.fs.mkdirs("abfss://< container_name>@<storage_account_name.dfs.core.windows.net/>< new_dir>")	mssparkutils.fs.mkdirs("file:/<new_dir>")
Varsayılan göl evi	"Dosyalar" veya "Tablolar" altındaki dizin: mssparkutils.fs.mkdirs("Dosyalar/<new_dir>")	mssparkutils.fs.mkdirs("abfss://< container_name>@<storage_account_name.dfs.core.windows.net/>< new_dir>")	mssparkutils.fs.mkdirs("file:/<new_dir>")

Dosyaları listeleme

Bir dizinin içeriğini listelemek için mssparkutils.fs.ls('Dizin yolunuz') kullanın. Örneğin:

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 

Dosya özelliklerini görüntüleme

Bu yöntem, dosya adı, dosya yolu, dosya boyutu ve dosyanın dizin mi yoksa dosya mı olduğu gibi dosya özelliklerini döndürür.

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

Yeni dizin oluşturma

Bu yöntem, belirtilen dizin mevcut değilse onu oluşturur ve gerekli tüm üst dizinleri de oluşturur.

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 

Dosyayı kopyalama

Bu yöntem bir dosya veya dizin kopyalar ve dosya sistemleri arasında kopyalama etkinliğini destekler.

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

Performanslı kopyalama dosyası

Bu yöntem, özellikle büyük hacimli verileri kopyalamak veya taşımak için daha hızlı bir yol sağlar.

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

Dosya içeriğini önizleme

Bu yöntem, belirtilen dosyanın ilk maxBytes baytlarına kadar UTF-8 ile kodlanmış bir dizi olarak geri döner.

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

Dosyayı taşıma

Bu yöntem bir dosyayı veya dizini taşır ve dosya sistemleri arasında taşımayı destekler.

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.

Dosya yazma

Bu yöntem, verilen dizeyi UTF-8 ile kodlanmış bir dosyaya yazar.

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

Dosyaya içerik ekleme

Bu yöntem, verilen dizeyi UTF-8 ile kodlanmış bir dosyaya ekler.

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

Not

API'yi mssparkutils.fs.append aynı dosyaya yazmak için döngü for içinde kullandığınızda, tekrarlayan yazımlar arasında yaklaşık 0,5 ila 1 saniyelik bir sleep ifade eklemenizi öneririz. mssparkutils.fs.append API'nin dahili flush işleyişi asenkrondur, bu yüzden kısa bir gecikme veri bütünlüğünü sağlar.

Dosya veya dizin silme

Bu yöntem bir dosyayı veya dizini kaldırır.

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

Dizini monte etme/çıkartma

Detaylı kullanım hakkında daha fazla bilgi için Dosya montajı ve çıkarma bölümünü inceleyin.

Dizüstü bilgisayar yardımcı programları

Not defterini çalıştırmak veya not defterinden bir değerle çıkmak için MSSparkUtils Not Defteri Yardımcı Programları'nı kullanın. Kullanılabilir yöntemlere genel bir bakış elde etmek için aşağıdaki komutu çalıştırın:

mssparkutils.notebook.help()

Çıktı:

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.

Not

Notebook yardımcı programları Apache Spark iş tanımlarına (SJD) uygulanmaz.

Not defterine başvurma

Bu yöntem bir not defterine başvurur ve çıkış değerini döndürür. İç içe işlev çağrılarını bir not defterinde, etkileşimli olarak veya bir işlem hattında çalıştırabilirsiniz. Atıfta bulunulan not defteri, bu işlevi çağıran not defterinin Spark havuzunda çalışır.

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

Örneğin:

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

Doku not defteri, çalışma alanı kimliğini belirterek birden çok çalışma alanında not defterlerine başvurmayı da destekler.

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

Referans çalıştırmasının anlık görüntü bağlantısını hücre çıktısında açabilirsiniz. Anlık görüntü, çalışan kodun sonuçlarını yakalar ve bir referans yürütmesinin hatalarını kolayca ayıklamanıza olanak tanır.

Not

• Çalışma alanları arasında referans alınan not defteri, 1.2 ve daha üstü çalışma zamanı sürümleriyle desteklenir.
• Notebook kaynakları altındaki dosyaları kullanıyorsanız, mssparkutils.nbResPath referans verilen notebook'u kullanarak etkileşimli çalıştırışla aynı klasörü işaret ettiğinden emin olun.

Birden çok not defterini paralel olarak çalıştırın

Önemli

Bu özellik önizleme aşamasındadır.

yöntemi mssparkutils.notebook.runMultiple() , birden çok not defterini paralel veya önceden tanımlanmış bir topolojik yapıyla çalıştırmanıza olanak tanır. API, mevcut Spark oturumu içinde yalıtılmış REPL örneklerinde (read-eval-print-loop) çalışan alt not defterlerini göndermek, sıraya almak ve izlemek için çok iş parçacıklı bir yapı kullanır. Referans verilen alt defterler, oturumun hesaplama kaynaklarını paylaşır.

ile mssparkutils.notebook.runMultiple()şunları yapabilirsiniz:

• Her birinin tamamlanmasını beklemeden birden çok not defterini aynı anda çalıştırın.

• Basit bir JSON biçimi kullanarak not defterleriniz için bağımlılıkları ve yürütme sırasını belirtin.

• Spark işlem kaynaklarının kullanımını iyileştirin ve Doku projelerinizin maliyetini azaltın.

• Her notebook çalıştırma kaydının anlık görüntülerini çıktıda görüntüleyin; notebook görevlerinizi kolayca hata ayıklayın ve izleyin.

• Her yönetici etkinliğinin çıkış değerini alın ve bunları aşağı akış görevlerinde kullanın.

Örneği ve ayrıntılı kullanımı bulmak için mssparkutils.notebook.help("runMultiple") komutunu çalıştırmayı da deneyebilirsiniz.

Aşağıda, bu yöntemi kullanarak not defterlerinin listesini paralel olarak çalıştırmanın basit bir örneği verilmiştir:

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

Kök not defterinin yürütme sonucu aşağıdaki gibidir:

Aşağıdaki örnek, mssparkutils.notebook.runMultiple() kullanarak topolojik yapıya sahip not defterlerinin nasıl çalıştırılacağını gösterir. Bir kod deneyimi aracılığıyla not defterlerini kolayca düzenlemek için bu yöntemi kullanın.

# 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ök not defterinin yürütme sonucu aşağıdaki gibidir:

Not

• Not defteri etkinlikleri veya eşzamanlı not defterleri için üst sınır, sürücü çekirdeği sayısıyla kısıtlanır. Örneğin, sekiz çekirdekli bir Medium düğüm sürücüsü aynı anda sekiz dizüstü bilgisayarı çalıştırabilir. Bu sınırlama, gönderilen her notebookun kendi REPL (read-eval-print-loop) örneğinde çalışması ve her örnek bir sürücü çekirdeği tüketmesi nedeniyle vardır.
• Kullanıcılar Spark havuzlarını daha büyük düğümlere ve dolayısıyla daha fazla sürücü çekirdeğine sahip yapılandırdıkça, varsayılan eşzamanlılık parametresi en yüksek eşzamanlılık ölçeğini otomatik olarak ölçeklendirmeyi desteklemek için 50 olarak ayarlanır. Daha büyük bir sürücü düğümü kullanıldığında bu parametri daha yüksek bir değere ayarlayabilseniz de, tek bir sürücü düğümünde çalışan eşzamanlı işlem sayısını artırmak genellikle doğrusal ölçeklenmez. Eşzamanlılığın artırılması, sürücü ve yürütücü kaynak çekişmesi nedeniyle verimliliğin azalmasına neden olabilir. Her çalışan dizüstü bilgisayar, sürücüdeki CPU ve belleği tüketen özel bir REPL örneğinde çalışır. Yüksek eşzamanlılık altında, bu tüketim özellikle uzun süreli iş yüklerinde sürücü kararsızlığı veya bellek dışı hatalar riskini artırabilir.
• REPL örneklerinin başlatılması ve çok sayıda notebook'un koordine edilmesi nedeniyle, her bir işte daha uzun yürütme süreleriyle karşılaşabilirsiniz. Sorun çıkarsa, notebookları birden fazla runMultiple çağrıya ayırmayı veya DAG parametresindeki eşzamanlılık alanını ayarlayarak eşzamanlılığı azaltmayı düşünün.
• Kısa ömürlü defterleri çalıştırdığınızda (örneğin, 5 saniyelik kod çalışma süresi), başlatma yükü baskın hale gelir. Hazırlık süresindeki değişkenlik, defterlerin örtüşme olasılığını azaltabilir ve dolayısıyla daha düşük gerçekleşen eşzamanlılığa yol açabilir. Bu durumlarda, küçük işlemleri bir veya birden fazla defterde birleştirmek daha optimal olabilir.
• Gönderim, kuyruğa alma ve izleme için çoklu iş parçacığı kullanılsa da her not defterinde çalışan kodun her yürütücüde çok iş parçacıklı olmadığını unutmayın. Defterler arasında kaynak paylaşımı yok. Her notebook sürecine toplam yürütücü kaynaklarının bir kısmı tahsis edilir. Bu tahsis, daha kısa süreli işlerin verimsiz çalışmasına ve daha uzun süreli işlerin kaynak mücadelesine yol açabilir.
• DAG’in tamamı için varsayılan zaman aşımı 12 saattir ve alt not defterlerindeki her hücre için varsayılan zaman aşımı 90 saniyedir. DAG parametresindeki timeoutInSeconds ve timeoutPerCellInSeconds alanlarını ayarlayarak zaman aşımını değiştirebilirsiniz. Eşzamanlılığı artırdıkça, olası kaynak çekişmesinin gereksiz zaman aşımlarına neden olmasını önlemek için timeoutPerCellInSeconds değerini artırmanız gerekebilir.

Not defterinden çıkış yap

Bu yöntem, belirtilen bir değerle bir defteri kapatır. İç içe işlev çağrılarını bir not defterinde, etkileşimli olarak veya bir işlem hattında çalıştırabilirsiniz.

• Bir not defterinden exit() işlevini etkileşimli olarak çağırdığınızda, Fabric not defteri bir özel durum oluşturur, sonraki hücreleri çalıştırmayı atlar ve Spark oturumunu canlı tutar.

• Exit() işlevini çağıran bir işlem hattında bir not defterini düzenlerseniz, not defteri etkinliği bir çıkış değeriyle döner, işlem hattı çalıştırmasını tamamlar ve Spark oturumunu durdurur. exit() işlevini bir try/catch bloğunun içine almayın; çünkü işlem hattının dönüş değerini alabilmesi için bu NotebookExit özel durumunun yayılması gerekir.

• Referans verilen bir defterde exit() fonksiyonunu çağırdığınızda, Fabric Spark referans verilen not defterinin daha fazla yürütülmesini durdurur ve ana defterde run() fonksiyonunu çağıran sonraki hücreleri çalıştırmaya devam eder. Örneğin: Notebook1 üç hücreye sahiptir ve ikinci hücrede exit () işlevini çağırır. Notebook2'nin beş hücresi vardır ve üçüncü hücrede run(notebook1) çağrıları vardır. Notebook2'yi çalıştırdığınızda, exit() işlevine basıldığında Notebook1 ikinci hücrede durur. Not Defteri2, dördüncü hücresini ve beşinci hücresini çalıştırmaya devam eder.

mssparkutils.notebook.exit("value string")

Örneğin:

Aşağıdaki iki hücreye sahip Örnek1 not defteri:

• Hücre 1, varsayılan değeri 10 olarak ayarlanmış bir giriş parametresi tanımlar.

• 2. hücre, çıkış değeri olarak giriş içeren not defterinden çıkar.

Sample1'i varsayılan değerlerle başka bir not defterinde çalıştırabilirsiniz:

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

Çıktı:

Notebook executed successfully with exit value 10

Sample1'i başka bir not defterinde çalıştırabilir ve giriş değerini 20 olarak ayarlayabilirsiniz:

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

Çıktı:

Notebook executed successfully with exit value 20

Kimlik Doğrulama Araçları

MSSparkUtils Credentials Utilities ile erişim tokenları alabilir ve Azure Key Vault'ta sırları yönetebilirsiniz.

Kullanılabilir yöntemlere genel bir bakış elde etmek için aşağıdaki komutu çalıştırın:

mssparkutils.credentials.help()

Çıktı:

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

Belirteç alma

getTokenbelirli bir hedef kitle ve isim için (isteğe bağlı) bir Microsoft Entra tokenı döndürür. Aşağıdaki listede şu anda kullanılabilir olan hedef kitle anahtarları gösterilmektedir:

• Depolama Hedef Kitle Kaynağı: storage
• Power BI Kaynağı:pbi
• Azure Key Vault Resource:keyvault
• Synapse RTA KQL DB Kaynak: kusto

Belirteci almak için aşağıdaki komutu çalıştırın:

mssparkutils.credentials.getToken('audience Key')

Kullanıcı kimlik bilgilerini kullanarak sır alın

getSecretkullanıcı kimlik bilgilerini kullanarak belirli bir Azure Key Vault uç noktası ve gizli adı için Azure Key Vault sırrı döndürür.

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

Dosya bağlama ve çıkarma

Doku, Microsoft Spark Yardımcı Programları paketinde aşağıdaki bağlama senaryolarını destekler. mount, unmount, getMountPath() ve mounts() API'lerini kullanarak uzak depolamayı (Azure Data Lake Storage 2. Nesil) tüm çalışma düğümlerine (sürücü düğümü ve işçi düğümleri) bağlayabilirsiniz. Depolama bağlama noktası gerçekleştikten sonra yerel dosya API'sini kullanarak verilere yerel dosya sisteminde depolanmış gibi erişin.

Azure Data Lake Storage 2. Nesil hesabı nasıl bağlanır

Aşağıdaki örnek, Azure Data Lake Storage 2. Nesil'nin nasıl monte edileceğini göstermektedir. Blob Depolama'yı monte etmek benzer şekilde çalışır.

Bu örnekte, storegen2 adlı bir Data Lake Storage 2. Nesil hesabınızın olduğu ve bu hesabın içinde mycontainer adlı bir kapsayıcı bulunup, not defteri Spark oturumunuza /test olarak bağlamak istediğiniz varsayılır.

mycontainer adlı konteyneri monte etmek için mssparkutils önce konteynere erişim izniniz olup olmadığını kontrol eder. Fabric, tetikleyici montaj işlemi için üç kimlik doğrulama yöntemini destekler: Microsoft Entra token (varsayılan ve önerilen), accountKey ve sastoken. Microsoft Entra token doğrulaması ve mevcut notebookutils API hakkında daha fazla bilgi için NotebookUtils dosya bağlama ve çıkarma (NotebookUtils file mount and unmount for Fabric sayfasına bakınız.

Paylaşılan erişim imza belirteçeği veya hesap anahtarı kullanarak montaj yapın

MSSparkUtils, bir hesap anahtarının veya Paylaşılan erişim imzası (SAS) belirtecinin hedefi bağlamak amacıyla parametre olarak açıkça geçirilmesini destekler.

Güvenlik nedeniyle, hesap anahtarlarını veya SAS belirteçlerini Azure Key Vault'ta depolamanızı öneririz (aşağıdaki ekran görüntüsünde gösterildiği gibi). Ardından bunları mssparkutils.credentials.getSecret API'sini kullanarak alabilirsiniz. Azure Key Vault hakkında daha fazla bilgi için Azure Key Vault yönetilen depolama hesabı anahtarları hakkında bölümüne bakın.

accountKey yöntemi için örnek kod:

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

Sastoken için örnek kod:

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

Not

Eğer mssparkutils kullanılamıyorsa, içe aktarmanız gerekebilir.

from notebookutils import mssparkutils

Bağlama parametreleri:

• fileCacheTimeout: Bloblar varsayılan olarak yerel geçici klasörde 120 saniye önbellek tutar. Bu süre boyunca blobfuse dosyanın güncel olup olmadığını denetlemez. Bu parametreyi varsayılan zaman aşımını değiştirmek için ayarlayın. Birden fazla istemci aynı anda dosyaları değiştirdiğinde, yerel ve uzak dosyalar arasında tutarsızlıkları önlemek için, önbellek süresini kısaltmanızı veya hatta 0'a çevirmenizi ve her zaman sunucudan en son dosyaları almanızı öneririz.
• timeout: Montaj operasyonunun zaman aşımı varsayılan olarak 120 saniyedir. Bu parametreyi varsayılan zaman aşımını değiştirmek için ayarlayın. Çok fazla yürütücü olduğunda veya montaj süresi bittiğinde, değeri artırmanızı öneririz.

Aşağıdaki gibi parametreleri kullanabilirsiniz:

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

Not

Güvenlik nedeniyle kimlik bilgilerini kodda depolamayın. Kimlik bilgilerinizi daha fazla korumak için, sır defter çıktısında sansürlenir. Daha fazla bilgi için bkz Gizli düzenleme.

Göl evi nasıl monte edilir?

Bir göl evi montajı için örnek kod:/test

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

Not

Bölgesel bir uç nokta kurmak desteklenmiyor. Fabric yalnızca onelake.dfs.fabric.microsoft.com genel uç noktasının bağlanmasını destekler.

Mount noktası altındaki dosyalara mssparkutils fs API kullanılarak erişim

Mount işleminin temel amacı, yerel dosya sistemi API'si kullanarak uzak depolama hesabında depolanan verilere erişmenizi sağlamaktır. mssparkutils fs API'sini, parametre olarak bağlı bir yol ile kullanarak verilere de erişebilirsiniz. Bu yol formatı biraz farklı.

Data Lake Storage 2. Nesil konteyner mycontainer'ı/test mount API'sini kullanarak monte ettiğinizi varsayalım. Yerel dosya sistemi API'si kullanarak verilere eriştiğinizde, yol formatı şöyledir:

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

mssparkutils fs API'sini kullanarak veriye erişmek istediğinizde, doğru yolu almak için getMountPath() kullanmanızı öneririz:

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

• Liste dizinleri:

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

• Dosya içeriğini okuma:

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

• Dizin oluşturma:

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

Bağlama noktası altındaki dosyalara yerel yol üzerinden erişme

Standart dosya sistemini kullanarak bağlama noktasındaki dosyaları kolayca okuyabilir ve yazabilirsiniz. Python örneği aşağıda verilmişti:

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

Mevcut bağlama noktalarını denetleme

Tüm mevcut bağlama noktası bilgilerini denetlemek için mssparkutils.fs.mounts() API'sini kullanabilirsiniz:

mssparkutils.fs.mounts()

Bağlama noktasını çıkarma

Bağlama noktanızı (/test bu örnekte) sökmek için aşağıdaki kodu kullanın:

mssparkutils.fs.unmount("/test")

Bilinen sınırlamalar

• Mevcut montaj iş seviyesinde bir konfigürasyondur. Bir bağlama noktasının mevcut olup olmadığını veya kullanılamadığını denetlemek için mounts API’sini kullanmanızı öneririz.

• Çıkarma mekanizması otomatik değildir. Uygulama çalıştırması tamamlandığında bağlama noktasını sökmek ve disk alanını serbest bırakmak için kodunuzda bir çıkarma API'sini açıkça çağırmanız gerekir. Aksi takdirde, uygulama çalıştırması tamamlandıktan sonra bağlama noktası düğümde hala var olur.

• Bir 1. Nesil Azure Data Lake Storage depolama hesabının bağlanması desteklenmez.

Lakehouse yardımcı programları

Modül, mssparkutils.lakehouse göl evi eşyalarının yönetimi için yardımcı programlar sağlar. Bu yardımcı araçlar, göl evi öğeleri oluşturmak, almak ve güncellemek için kolay bir şekilde silmeyi sağlar.

Not

Lakehouse API'leri yalnızca Runtime sürüm 1.2 veya daha sonrasında desteklenmektedir.

Yöntemlere genel bakış

Modülde aşağıdaki yöntemler mevcuttur 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]

Kullanım örnekleri

Bu yöntemleri etkili kullanmak için aşağıdaki kullanım örneklerini göz önünde bulundurun:

Bir göl evi eşyası oluşturmak

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

Bir göl evi eşyasını almak

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

Bir göl evi ürünü güncelleme

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

Bir göl evi eşyasını silmek

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

Göl evi eşyalarının listelenmesi

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

Ek bilgi

Her yöntem ve parametreleri hakkında daha ayrıntılı bilgi için fonksiyonu mssparkutils.lakehouse.help("methodName") kullanın.

MSSparkUtils'in Lakehouse altyapılarını kullanarak, lakehouse öğelerinizi daha verimli yönetebilir ve bu yönetimi Fabric boru hatlarınıza entegre ederek genel veri yönetimi deneyiminizi geliştirebilirsiniz.

Bu altyapıları keşfedin ve sorunsuz göl evi eşya yönetimi için Fabric iş akışlarınıza dahil edin.

Çalışma zamanı yardımcı programları

Oturum bağlam bilgilerini gösterme

mssparkutils.runtime.context kullanarak, not defteri adı, varsayılan lakehouse, çalışma alanı bilgileri, bunun bir işlem hattı çalıştırması olup olmadığı ve daha fazlası dahil olmak üzere geçerli canlı oturumun bağlam bilgilerini alabilirsiniz.

mssparkutils.runtime.context

Not

mssparkutils.envFabric'te resmi olarak desteklenmiyor. Alternatif olarak kullanın notebookutils.runtime.context .

Bilinen sorun

1.2'den daha eski bir çalışma zamanı sürümü kullandığınızda ve çalıştırıldığında mssparkutils.help(), listelenen fabricClient, depo ve çalışma alanı API'leri şu anda desteklenmiyor.

İlgili içerik

• Kitaplık yönetimi