Fabric 用 Microsoft Spark Utilities (MSSparkUtils)

Microsoft Spark Utilities(MSSparkUtils)は、一般的な作業を簡単に実行できる組み込みパッケージです。 MSSparkUtils を使用して、ファイル システムの操作、環境変数の取得、ノートブックの連結、シークレットの操作を行います。 MSSparkUtilsパッケージはPySpark(Python)、Scala、SparkRノートブック、Fabricパイプラインで利用可能です。

ファイルシステム ユーティリティ

mssparkutils.fsは、Azure Data Lake Storage Gen2やAzure Blob Storageを含む様々なファイルシステムで作業するためのユーティリティを提供しています。 Azure Data Lake Storage Gen2 および Azure Blob Storage へのアクセスを適切に構成するようにしてください。

次のコマンドを実行して、使用可能なメソッドの概要を取得します。

from notebookutils import mssparkutils
mssparkutils.fs.help()

出力

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.

MSSparkUtil でファイル システムにアクセスする方法は、Spark API と同じです。 例えば mssparkuitls.fs.mkdirs() やLakehouseの使用例を挙げてみましょう:

ファイルの一覧表示

ディレクトリの内容を一覧表示するには、mssparkutils.fs.ls('ディレクトリ パス') を使用します。 次に例を示します：

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 

ファイルのプロパティを表示します

このメソッドはファイル名、ファイルパス、ファイルサイズ、ディレクトリかファイルかなどのファイルプロパティを返します。

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

新しいディレクトリの作成

このメソッドは、指定されたディレクトリが存在しない場合に作成し、必要な親ディレクトリを作成します。

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 

ファイルのコピー

このメソッドでは、ファイルまたはディレクトリをコピーし、ファイル システム間のコピー アクティビティをサポートします。

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

パフォーマンスの高いコピー ファイル

このメソッドでは、ファイル (特に大量のデータ) を迅速にコピーまたは移動する方法を提供します。

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

ファイル コンテンツのプレビュー

この方法は、指定されたファイルの最初の maxBytes バイトまでUTF-8でエンコードされた文字列として返します。

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

ファイルの移動

このメソッドでは、ファイルまたはディレクトリを移動し、ファイル システム間の移動をサポートします。

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.

ファイルの書き込み

このメソッドでは、指定した文字列を 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

ファイルへのコンテンツの追加

このメソッドでは、指定した文字列を 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

ファイルまたはディレクトリの削除

このメソッドでは、ファイルまたはディレクトリを削除します。

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

ディレクトリのマウント、マウント解除

詳細な使用方法については、 ファイルマウントおよびアンマウントをご覧ください。

Notebook のユーティリティ

MSSparkUtils のNotebook ユーティリティでは、Notebookを実行することや、終了値の指定付きでNotebookを終了させることができます。 次のコマンドを実行して、使用可能なメソッドの概要を取得します。

mssparkutils.notebook.help()

出力:

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.

ノートを参照する

このメソッドでは、Notebookを参照し、その終了値を返します。 入れ子になった関数呼び出しは、ノートブックで対話的に、またはパイプライン内で実行できます。 参照されたNotebookは、この関数を呼び出すNotebookの Spark プールで実行されます。

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

次に例を示します：

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

Fabric Notebookでは、ワークスペース ID を指定することで、複数のワークスペース間でのNotebookの参照もサポートしています。

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

セルの出力部分で参照実行のスナップショットリンクを開くことができます。 このスナップショットでは、コードの実行結果をキャプチャし、参照の実行を簡単にデバッグできるようにします。

参照で複数のノートブックを並列に実行する

mssparkutils.notebook.runMultiple() メソッドを使用すると、複数のNotebookを並列で、または定義済みのトポロジ構造で実行できます。 このAPIはマルチスレッド実装を用いて、既存のSparkセッション内で分離されたREPLインスタンス(read-eval-print-loop)で実行される子ノートブックを送信、キュー、監視します。 参照された子ノートブックはセッションの計算リソースを共有します。

mssparkutils.notebook.runMultiple() を使用すると、以下のことができます。

• 各ノートブックが完了するのを待たずに、複数を同時に実行します。

• 単純な JSON 形式を使用して、Notebookの依存関係と実行順序を指定します。

• Spark コンピューティング リソースの使用を最適化し、Fabric プロジェクトのコストを削減します。

• 出力で各ノートブックの実行記録のスナップショットを確認でき、ノートブックのタスクを便利にデバッグ・監視できます。

• 各エグゼクティブ アクティビティの終了値を取得し、ダウンストリーム タスクで使用します。

mssparkutils.notebook.help("runMultiple") を実行して、例と詳細な使用方法を見つけることもできます。

このメソッドを使用してノートブックの一覧を並列実行する方法についての簡単な例を次に示します。

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

ルート Notebookからの実行結果は次のとおりです。

以下の例は、 mssparkutils.notebook.runMultiple()を用いてトポロジカル構造を持つノートブックを動かしている様子を示しています。 コード エクスペリエンスを使用してNotebookを簡単にオーケストレーションするには、このメソッドを使用します。

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

ルート Notebookからの実行結果は次のとおりです。

ノートブックを終了する

このメソッドでは、値を指定してNotebookを終了します。 入れ子になった関数呼び出しは、ノートブックで対話的に、またはパイプライン内で実行できます。

• exit() 関数をNotebookから対話形式で呼び出すと、Fabric Notebookによって例外がスローされて、以後に続くセルの実行がスキップされ、Spark セッションは維持されます。

• exit() 関数を呼び出すパイプラインでNotebookのオーケストレーションを行うと、そのNotebook アクティビティは終了して終了値を返し、パイプラインの実行が完了し、Spark セッションが停止します。 try/catchの周りに exit() 関数を囲まないでください。このNotebookExit Exceptionはパイプラインが戻り値を取得するために伝播しなければなりません。

• 参照中のノートブックでexit()関数を呼び出すと、Fabric Sparkは参照されたノートブックの実行を停止し、run()関数を呼び出すメインノートブックの次のセルを実行し続けます。 たとえば、Notebook1 には 3 つのセルがあり、2 番目のセルで exit() 関数が呼び出されるとします。 Notebook2 には 5 つのセルがあり、3 番目のセルで run(notebook1) が呼び出されるとします。 Notebook2 を実行すると、Notebook1 は、2 番目のセルで exit() 関数に到達して停止します。 Notebook2 は 4 番目のセルと 5 番目のセルを引き続き実行します。

mssparkutils.notebook.exit("value string")

次に例を示します：

Sample1 Notebookには、以下の 2 つのセルがあります。

• セル 1 では、input パラメーターがデフォルト値 10 で定義されます。

• セル 2 はノートブックを終了し、終了値として input が用いられます。

デフォルト値を使用して、別のNotebookで Sample1 を実行できます。

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

出力:

Notebook executed successfully with exit value 10

別のNotebookで Sample1 を実行して、入力値を 20 に設定できます。

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

出力:

Notebook executed successfully with exit value 20

資格情報ユーティリティ

MSSparkUtils Credentials Utilitiesを使ってアクセストークンを取得し、Azure Key Vault内の秘密管理が可能です。

次のコマンドを実行して、使用可能なメソッドの概要を取得します。

mssparkutils.credentials.help()

出力:

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

トークンを取得する

getToken特定のオーディエンスと名前(任意)に対してMicrosoft Entraトークンを返します。 次の一覧に、現在利用可能な対象ユーザー キーを示します。

• ストレージオーディエンスリソース: storage
• Power BIリソース:pbi
• Azure Key Vault Resource:keyvault
• Synapse RTA KQL DB Resource: kusto

トークンを取得するには、以下のコマンドを実行します。

mssparkutils.credentials.getToken('audience Key')

ユーザー認証情報を使って秘密情報を取得する

getSecretユーザー認証情報を用いて、特定のAzure Key Vaultエンドポイントと秘密名に対してAzure Key Vaultの秘密を返します。

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

ファイルのマウントとマウント解除

Fabric では、Microsoft Spark Utilities パッケージの次のマウント シナリオをサポートしています。 mount、unmount、getMountPath()、mounts() APIを使って、リモートストレージ(Azure Data Lake Storage Gen2)をすべての動作ノード(ドライバーノードとワーカーノード)に接続できます。 ストレージ マウント ポイントを確立した後は、データがローカル ファイル システムに格納されているかのように、ローカル ファイル API を使用してアクセスできます。

Azure Data Lake Storage Gen2アカウントのマウント方法

以下の例は、Azure Data Lake Storage Gen2のマウント方法を示しています。 Blob Storage のマウントも同様に機能します。

この例では、storegen2 という名前の 1 つの Data Lake Storage Gen2 アカウントがあり、そのアカウントに mycontainer という名前のコンテナーが 1 つあり、それをNotebook Spark セッション内で /test にマウントしようとしていることを前提としています。

mycontainerという名前のコンテナをマウントする際、mssparkutilsはまずコンテナへのアクセス権限があるかどうかを確認します。 Fabricはトリガーマウント操作の3つの認証方法をサポートしています:Microsoft Entraトークン(デフォルトおよび推奨)、accountKey、そしてsastokenです。 Microsoft Entra トークン認証および現行の notebookutils API の詳細については、Fabric 向けの NotebookUtils のファイルのマウントとマウント解除をご覧ください。

共有アクセス署名トークンまたはアカウントキーを使ってマウントします

MSSparkUtils では、ターゲットをマウントする際に、アカウント キーまたは Shared Access Signature (SAS) トークンをパラメーターとして明示的に渡すことができます。

セキュリティ上の理由から、アカウント キーまたは SAS トークンを Azure Key Vault に保存することをお勧めします (次のスクリーンショットを参照)。 以後は、mssparkutils.credentials.getSecret API でそれらのシークレットを取得できます。 Azure Key Vault の詳細については、「Azure Key Vault のマネージド ストレージ アカウント キーについて」をご覧ください。

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

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

マウント パラメーター:

• fileCacheTimeout: Blobsはデフォルトでローカルのtempフォルダに120秒キャッシュされます。 この間、blobfuse はファイルが最新かどうかを確認しません。 このパラメータを設定してデフォルトのタイムアウトを変更してください。 複数のクライアントが同時にファイルを変更する場合、ローカルファイルとリモートファイルの不整合を避けるために、キャッシュ時間を短縮するか、0に設定することを推奨し、常にサーバーから最新のファイルを取得することをお勧めします。
• timeout: マウント操作のタイムアウトはデフォルトで120秒です。 このパラメータを設定してデフォルトのタイムアウトを変更してください。 エグゼキューターが多すぎる場合やマウントタイムアウトの場合は、価値を上げることを推奨します。

これらのパラメーターの使用例を示します。

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

レイクハウスをマウントする方法

/testに湖畔ハウスを設置するためのサンプルコード:

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

mssparkutils fs APIを使ってマウントポイントの下のファイルにアクセスする

マウント操作の主な目的は、ローカルファイルシステムのAPIを使ってリモートストレージアカウントに保存されたデータにアクセスできるようにすることです。 また、mssparkutils fs API でも、マウントされたパスをパラメーターで指定してデータにアクセスすることができます。 このパス形式は少し異なります。

マウントAPIを使ってData Lake Storage Gen2コンテナのmycontainerを/testにマウントしたと仮定します。 ローカルファイルシステムAPIを使ってデータにアクセスする場合、パス形式は以下の通りです:

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

mssparkutils fs APIを使ってデータにアクセスしたい場合は、正確なパスを得るためにgetMountPath()を使うことをお勧めします。

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

• ディレクトリを一覧表示する。

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

• ファイルの内容を読み取ります。

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

• ディレクトリを作成します。

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

ローカル パスを使用してマウント ポイント内のファイルにアクセスする

マウント ポイント内のファイルは、標準のファイル システムを使って簡単に読み書きできます。 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"))

既存のマウント ポイントをチェックする方法

mssparkutils.fs.mounts() API を使用すると、既にあるすべてのマウント ポイントの情報を確認できます。

mssparkutils.fs.mounts()

マウント ポイントをマウント解除する方法

マウント ポイント (この例では /test) をマウント解除するには、次のようなコードを実行します。

mssparkutils.fs.unmount("/test")

既知の制限事項

• 現在のマウントはジョブレベルの設定です。 マウントポイントの存在や利用不可を確認するために マウントAPIを使う ことをお勧めします。

• マウント解除のメカニズムは自動的には実行されません。 アプリケーションの実行が完了したときに、マウント ポイントをマウント解除してディスク領域を解放するには、コードでマウント解除 API を明示的に呼び出す必要があります。 それ以外の場合、マウント ポイントは、アプリケーションの実行が完了した後もノードに存在します。

• Azure Data Lake Storage Gen1ストレージアカウントのマウントはサポートされていません。

レイクハウスのユーティリティ

mssparkutils.lakehouseモジュールは湖畔のアイテム管理のためのユーティリティを提供します。 これらのユーティリティは、レイクハウスアイテムの作成、取得、更新、削除を簡単にします。

メソッドの概要

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]

使用例

これらの手法を効果的に活用するために、以下の使用例を検討してください。

湖の家アイテムの作成

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

湖畔の家のアイテムの取り出し

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

湖畔の家のアイテムの更新

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

湖の家のアイテムを削除する

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

湖畔の家のアイテム一覧

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

追加情報

各メソッドとそのパラメータの詳細については、 mssparkutils.lakehouse.help("methodName") 関数をご利用ください。

MSSparkUtilsのLakehouseユーティリティを活用することで、Lakehouseのアイテムをより効率的に管理し、その管理をFabricパイプラインに統合することで、全体的なデータ管理体験を向上させます。

これらのユーティリティを探索し、Fabricのワークフローに組み込んで、シームレスなレイクハウスアイテム管理を実現しましょう。

ランタイム ユーティリティ

セッション コンテキスト情報を表示する

mssparkutils.runtime.contextを使うことで、現在のライブセッションのコンテキスト情報(ノートブック名、デフォルトのレイクハウス、ワークスペース情報、パイプライン実行の有無など)を取得できます。

mssparkutils.runtime.context

既知の問題

1.2より後のランタイムバージョンを使い、 mssparkutils.help()実行すると、上記の fabricClient、 ウェアハウス、 ワークスペース APIは現在サポートされていません。

関連するコンテンツ

• ライブラリ管理