Fabric 用 Microsoft Spark Utilities (MSSparkUtils)

Microsoft Spark Utilities (MSSparkUtils) は、一般的なタスクをより簡単に実行できるようにする組み込みパッケージです。 MSSparkUtils を使用すると、ファイル システムを操作し、環境変数を取得し、Notebookをまとめてチェーン化し、シークレットを操作できます。 MSSparkUtils パッケージは、PySpark (Python)、Scala、SparkR Notebook、Fabric パイプラインで利用できます。

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

mssparkutils.fs には、Azure Data Lake Storage (ADLS 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() で Fabric 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 でエンコードされた文字列として返します。

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

ファイルの移動

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

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): 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.

Notebookの参照

このメソッドでは、Notebookを参照し、その終了値を返します。 関数呼び出しの入れ子は、対話形式またはパイプラインで、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")

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

複数のNotebook参照を並列で実行する

mssparkutils.notebook.runMultiple() メソッドを使用すると、複数のNotebookを並列で、または定義済みのトポロジ構造で実行できます。 この API は、Spark セッション内でのマルチスレッド実装メカニズムを使用しています。つまり、コンピューティング リソースは参照Notebookの実行によって共有されます。

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

• 各Notebookが完了するのを待たずに、複数のNotebookを同時に実行します。

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

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

• 出力内の各Notebook実行レコードのスナップショットを表示し、Notebook タスクを容易にデバッグ/監視します。

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

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

このメソッドを使用して一連のNotebookを並列で実行する簡単な例を次に示します。

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

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

次に、mssparkutils.notebook.runMultiple()を使用してトポロジ構造を持つ複数のNotebookを実行する例を示します。 コード エクスペリエンスを使用して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, default to 50
}
mssparkutils.notebook.runMultiple(DAG, {"displayDAGViaGraphviz": False})

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

Notebookを終了する

このメソッドでは、値を指定してNotebookを終了します。 関数呼び出しの入れ子は、対話形式またはパイプラインで、Notebookで実行できます。

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

• exit() 関数を呼び出すパイプラインでNotebookのオーケストレーションを行うと、そのNotebook アクティビティは終了して終了値を返し、パイプラインの実行が完了し、Spark セッションが停止します。

• 参照されたNotebook内で exit() 関数を呼び出すと、Fabric Spark では、参照されたNotebookの実行をその時点で停止し、run() 関数を呼び出すメイン Notebook内の次のセルから実行を継続します。 たとえば、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 では、Notebookが終了し、終了値として 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 の資格情報ユーティリティを使用すると、アクセス トークンを取得し、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 リソース: "keyvault"
• Synapse RTA KQL DB リソース: "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 を使用して、リモート ストレージ (ADLS Gen2) を、すべての作業ノード (ドライバー ノードとワーカー ノード) にアタッチできます。 ストレージ マウント ポイントを確立した後は、データがローカル ファイル システムに格納されているかのように、ローカル ファイル API を使用してアクセスできます。

ADLS Gen2 アカウントをマウントする方法

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

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

mycontainer という名前のコンテナーをマウントするには、まず、このコンテナーにアクセスする権限が作業者に付与されているかどうかを mssparkutils がチェックする必要があります。 現在、Fabric では、トリガー マウント操作の認証方法として、accountKey と sastoken の 2 つをサポートしています。

Shared Access Signature トークンまたはアカウント キーを使ってマウントする

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: BLOB は、ローカルの一時フォルダーにデフォルトで 120 秒間キャッシュされます。 その期間内には、blobfuse はファイルが最新であるかどうかを確認しません。 必要に応じて、パラメーターの指定によりデフォルトのタイムアウト時間を変更できます。 複数のクライアントから同時にファイルの変更操作が行われる場合は、ローカル ファイルとリモート ファイルとの間に不整合が発生するのを避けるために、キャッシュ時間を短縮するか 0 に設定し、常にサーバーから最新のファイルが取得されるようにすることをお勧めします。
• timeout: マウント操作のタイムアウトは、デフォルトで 120 秒です。 必要に応じて、パラメーターの指定によりデフォルトのタイムアウト時間を変更できます。 実行者が多すぎる場合や、マウントのタイムアウトが発生する場合は、値を増やすことをお勧めします。

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

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

Lakehouseをマウントする方法

Lakehouseを /test にマウントするサンプル コードを次に示します。

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

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

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

たとえば、Data Lake Storage Gen2 コンテナー mycontainer が、mount API で /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")

既知の制限事項

• 現在、マウントはジョブ レベルで構成されています。マウント ポイントが存在するか、利用不可であるかを mounts API で確認することをお勧めします。

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

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

Lakehouse ユーティリティ

mssparkutils.lakehouse は、Lakehouse アーティファクトの管理用に特別に調整されたユーティリティを提供します。 これらのユーティリティを使用すると、ユーザーは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]

使用例

これらのメソッドを効果的に利用するには、次の使用例を検討してください。

Lakehouse アーティファクトの作成

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

Lakehouse アーティファクトの取得

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

Lakehouse アーティファクトの更新

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

Lakehouse アーティファクトの削除

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

Lakehouse アーティファクトの一覧表示

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

追加情報

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

MSSparkUtils の Lakehouse ユーティリティを使用すると、Lakehouse アーティファクトの管理がより効率的になり、Fabric パイプラインに統合され、全体的なデータ管理エクスペリエンスが向上します。

これらのユーティリティを自由に探索し、それらを Fabric ワークフローに組み込んで、シームレスなLakehouse アーティファクト管理を行うことができます。

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

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

mssparkutils.runtime.context を使用して、現在のライブ セッションのコンテキスト情報 (Notebook名、デフォルトのLakehouse、ワークスペース情報、パイプライン実行であるかどうかなど) を取得できます。

mssparkutils.runtime.context

既知の問題

ランタイム バージョン 1.2 以降を使用して mssparkutils.help() を実行する場合、一覧表示されている fabricClient、Warehouse、ワークスペース API は、現在サポートされていませんが、今後使用できるようになります。

関連するコンテンツ

• ライブラリ管理