Fabric용 NotebookUtils(이Notebook MSSparkUtils)

아티클
12/13/2024

Notebook 유틸리티(NotebookUtils)는 Fabric Notebook에서 일반적인 작업을 쉽게 수행할 수 있도록 도와주는 기본 제공 패키지입니다. NotebookUtils를 사용하여 파일 시스템 작업을 하고, 환경 변수를 가져오고, Notebook을 서로 연결하고, 비밀을 사용할 수 있습니다. NotebookUtils 패키지는 PySpark(Python) Scala, SparkR Notebook 및 Fabric 파이프라인에서 사용할 수 있습니다.

참고 항목

MsSparkUtils는 공식적으로 NotebookUtils이름이 바뀝니다. 기존 코드는 이전 버전과 호환성을 유지하며 호환성이 깨지는 변경은 발생하지 않습니다. 새로운 기능에 대한 지속적인 지원과 액세스를 보장하기 위해 Notebookutils로 업그레이드하는 것을 강력히 권장합니다. Mssparkutils 네임스페이스는 나중에 사용 중지됩니다.
NotebookUtils는 Spark 3.4(런타임 v1.2) 이상에서 작동하도록 설계되었습니다. 모든 새로운 기능 및 업데이트는 앞으로 Notebookutils 네임스페이스에서만 지원됩니다.

파일 시스템 유틸리티

notebookutils.fs 는 ADLS(Azure Data Lake Storage) Gen2 및 Azure Blob Storage를 비롯한 다양한 파일 시스템으로 작업하기 위한 유틸리티를 제공합니다. Azure Data Lake Storage Gen2 및 Azure Blob Storage에 대한 액세스를 적절하게 구성해야 합니다.

사용 가능한 메서드에 관한 개요를 가져오려면 다음 명령을 실행합니다.

notebookutils.fs.help()

출력

notebookutils.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
fastcp(from: String, to: String, recurse: Boolean = true): Boolean -> [Preview] Copies a file or directory via azcopy, possibly across FileSystems
mv(from: String, to: String, createPath: Boolean = false, overwrite: 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 notebookutils.fs.help("methodName") for more info about a method.

NotebookUtils는 Spark API와 동일한 방식으로 파일 시스템에서 작동합니다. Notebookutils.fs.mkdirs() 및 Fabric Lakehouse 사용량을 예로 들 수 있습니다.

사용법	HDFS 루트의 상대 경로	ABFS 파일 시스템의 절대 경로	드라이버 노드의 로컬 파일 시스템에 대한 절대 경로
기본이 아닌 lakehouse	지원되지 않음	notebookutils.fs.mkdirs("abfss://<container_name>@<storage_account_name>.dfs.core.windows.net/<new_dir>")	notebookutils.fs.mkdirs("file:/<new_dir>")
기본 lakehouse	"파일" 또는 "테이블" 아래의 디렉터리: notebookutils.fs.mkdirs("Files/<new_dir>")	notebookutils.fs.mkdirs("abfss://<container_name>@<storage_account_name>.dfs.core.windows.net/<new_dir>")	notebookutils.fs.mkdirs("file:/<new_dir>")

파일 나열

디렉터리의 콘텐츠를 나열하려면 notebookutils.fs.ls('디렉터리 경로')를 사용합니다. 예시:

notebookutils.fs.ls("Files/tmp") # The relatvie path may work with different base path, details in below 
notebookutils.fs.ls("abfss://<container_name>@<storage_account_name>.dfs.core.windows.net/<path>")  # The absolute path, like: ABFS file system
notebookutils.fs.ls("file:/tmp")  # The full path of the local file system of driver node

notebookutils.fs.ls() API는 Notebook 유형에 따라 상대 경로를 사용할 때 다르게 동작합니다.

"Spark Notebook": 상대 경로는 기본 Lakehouse의 "ABFSS" 경로를 기준으로 합니다. 예를 들어 notebookutils.fs.ls("Files") 기본 Lakehouse의 Files 디렉터리를 가리킵니다.

예시:
```
notebookutils.fs.ls("Files/sample_datasets/public_holidays.parquet")
```
파이썬 노트북: 상대 경로는 로컬 파일 시스템의 작업 디렉터리를 기준으로 하며, 기본적으로 /home/trusted-service-user/work 디렉터리에 있습니다. 따라서 상대 경로 notebookutils.fs.ls("/lakehouse/default/Files") 대신 전체 경로를 사용하여 기본 Lakehouse의 Files 디렉터리에 액세스해야 합니다.

예시:
```
notebookutils.fs.ls("/lakehouse/default/Files/sample_datasets/public_holidays.parquet")
```

파일 속성 보기

이 메서드는 파일 이름, 파일 경로, 파일 크기, 디렉토리 및 파일 여부 등의 파일 속성을 반환합니다.

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

새 디렉터리 만들기

이 메서드는 지정된 디렉터리가 없는 경우 해당 디렉터리를 만들고 필요한 부모 디렉터리를 만듭니다.

notebookutils.fs.mkdirs('new directory name')  
notebookutils.fs.mkdirs("Files/<new_dir>")  # works with the default lakehouse files using relative path 
notebookutils.fs.ls("abfss://<container_name>@<storage_account_name>.dfs.core.windows.net/<new_dir>")  # based on ABFS file system 
notebookutils.fs.ls("file:/<new_dir>")  # based on local file system of driver node

파일 복사

이 메서드는 파일 또는 디렉터리를 복사하고 파일 시스템에서 복사 작업을 지원합니다.

notebookutils.fs.cp('source file or directory', 'destination file or directory', True)# Set the third parameter as True to copy all files and directories recursively

참고 항목

OneLake 바로 가기제한으로 인해 사용하여 S3/GCS 형식 바로 가기에서 데이터를 복사해야 하는 경우 abfss 경로 대신 탑재된 경로를 사용하는 것이 좋습니다.

성능 복사 파일

이 메서드는 특히 대용량 데이터 볼륨을 처리할 때 파일을 복사하거나 이동하는 보다 효율적인 방법을 제공합니다. Fabric에서 성능을 향상시키려면 기존 fastcp메서드드 대신 cp를 활용하는 것이 좋습니다.

참고 항목

notebookutils.fs.fastcp() 는 지역 간 OneLake의 파일 복사를 지원하지 않습니다. 이 경우 대신 사용할 notebookutils.fs.cp() 수 있습니다.
OneLake 바로 가기의 제한으로 인해 S3/GCS 형식의 바로 가기에서 데이터를 복사해야 할 때 를 사용할 경우, abfss 경로 대신 탑재된 경로를 사용하는 것이 좋습니다.

notebookutils.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로 인코딩된 문자열로 반환합니다.

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

파일 이동

이 메서드는 파일 또는 디렉터리를 이동하고 파일 시스템 간 이동을 지원합니다.

notebookutils.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
notebookutils.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으로 인코딩된 파일에 씁니다.

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

파일에 콘텐츠 추가

이 메서드는 지정된 문자열을 UTF-8로 인코딩된 파일에 추가합니다.

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

참고 항목

notebookutils.fs.append() 원 notebookutils.fs.put() 자성 보장이 부족하여 동일한 파일에 대한 동시 쓰기를 지원하지 않습니다.
for 루프에서 notebookutils.fs.append API를 사용하여 동일한 파일에 쓰는 경우 되풀이 쓰기 사이에 약 0.5s ~ 1s의 sleep 문을 추가하는 것이 좋습니다. 이 권장 사항은 notebookutils.fs.append API의 내부 flush 작업이 비동기적이므로 짧은 지연이 데이터 무결성을 보장하는 데 도움이 됩니다.

파일 또는 디렉터리 삭제

이 메서드는 파일 또는 디렉토리를 제거합니다.

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

탑재/분리 디렉터리

파일 탑재 및 분리의 자세한 사용에 대한 자세한 정보를 알아봅니다.

Notebook 유틸리티

Notebook 유틸리티를 사용하여 Notebook을 실행하거나 값이 있는 Notebook을 종료합니다. 사용 가능한 메서드에 관한 개요를 가져오려면 다음 명령을 실행합니다.

notebookutils.notebook.help()

출력:


The notebook module.

exit(value: String): void -> This method lets you exit a notebook with a value.
run(path: String, timeoutSeconds: int, arguments: Map, workspace: String): String -> This method runs a notebook and returns its exit value.
runMultiple(DAG: Any): Map[String, MsNotebookRunResult] -> [Preview] Runs multiple notebooks concurrently with support for dependency relationships.
validateDAG(DAG: Any): Boolean -> [Preview] This method check if the DAG is correctly defined.

[Preview] Below methods are only support Fabric Notebook.
create(name: String, description: String = "", content: String = "", defaultLakehouse: String = "", defaultLakehouseWorkspace: String = "", workspaceId: String = ""): Artifact -> Create a new Notebook.
get(name: String, workspaceId: String = ""): Artifact -> Get a Notebook by name or id.
update(name: String, newName: String, description: String = "", workspaceId: String = ""): Artifact -> Update a Artifact by name.
delete(name: String, workspaceId: String = ""): Boolean -> Delete a Notebook by name.
list(workspaceId: String = "", maxResults: Int = 1000): Array[Artifact] -> List all Notebooks in the workspace.
updateDefinition(name: String, content: String = "", defaultLakehouse: String = "", defaultLakehouseWorkspace: String = "", workspaceId: String = "") -> Update the definition of a Notebook.

Use notebookutils.notebook.help("methodName") for more info about a method.

참고 항목

Notebook 유틸리티는 Apache Spark SJD(작업 정의)에 적용되지 않습니다.

Notebook 참조

이 메서드는 노트북을 참조하고 그 종료 값을 반환합니다. Notebook 또는 파이프라인에서 중첩 함수 호출을 실행할 수 있습니다. 참조 중인 Notebook은 Notebook이 이 함수를 호출하는 Spark 풀에서 실행됩니다.

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

예시:

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

Fabric Notebook은 작업 영역 ID를 지정하여 여러 작업 영역에서 Notebook 참조를 지원합니다.

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

셀 출력에서 참조 실행의 스냅샷 링크를 열 수 있습니다. 스냅샷은 코드 실행 결과를 캡처하고 참조 실행을 쉽게 디버그할 수 있도록 합니다.

참고 항목

작업 영역 간 참조 Notebook은 런타임 버전 1.2 이상에서 지원됩니다.
Notebook 리소스 아래의 파일을 사용하는 경우 참조된 Notebook에서 notebookutils.nbResPath를 사용해 대화형 실행과 동일한 폴더를 가리키는지 확인합니다.

참조는 여러 Notebook을 병렬로 실행합니다.

Important

이 기능은 프리뷰로 제공됩니다.

notebookutils.notebook.runMultiple() 메서드를 사용하면 여러 Notebook을 병렬로 실행하거나 미리 정의된 토폴로지 구조로 실행할 수 있습니다. API는 Spark 세션 내에서 다중 스레드 구현 메커니즘을 사용합니다. 즉, 참조 노트북 실행이 계산 자원을 공유합니다.

notebookutils.notebook.runMultiple()로 다음을 수행할 수 있습니다.

각 Notebook이 완료되는 것을 기다리지 않고 동시에 여러 Notebook을 실행합니다.
간단한 JSON 형식을 사용하여 Notebook에 대한 종속성 및 실행 순서를 지정합니다.
Spark 컴퓨팅 리소스 사용을 최적화하고 Fabric 프로젝트의 비용을 절감합니다.
출력에서 각 Notebook 실행 레코드의 스냅샷을 보고 Notebook 작업을 편리하게 디버그/모니터링합니다.
각 임원 활동의 종료 값을 가져와 다운스트림 작업에 사용합니다.

Notebookutils.notebook.help("runMultiple")를 실행하여 예제 및 자세한 사용량을 찾을 수도 있습니다.

다음은 이 방법을 사용하여 Notebook 목록을 병렬로 실행하는 간단한 예제입니다.


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

루트 Notebook의 실행 결과는 다음과 같습니다.

다음은 notebookutils.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
}
notebookutils.notebook.runMultiple(DAG, {"displayDAGViaGraphviz": False})

루트 Notebook의 실행 결과는 다음과 같습니다.

또한 DAG가 올바르게 정의되었는지 확인하는 메서드를 제공합니다.

notebookutils.notebook.validateDAG(DAG)

참고 항목

여러 Notebook 실행의 병렬 처리 수준은 Spark 세션의 사용 가능한 총 컴퓨팅 리소스로 제한됩니다.
Notebook 활동 또는 동시 Notebook의 상한은 50입니다. 이 제한을 초과하면 높은 컴퓨팅 리소스 사용량으로 인해 안정성 및 성능 문제가 발생할 수 있습니다. 문제가 발생하는 경우 DAG 매개 변수에서 runMultiple 필드를 조정하여 Notebook을 여러 호출로 분리하거나 동시성을 줄이는 것이 좋습니다.
전체 DAG의 기본 제한 시간은 12시간이며 자식 Notebook의 각 셀에 대한 기본 제한 시간은 90초입니다. DAG 매개 변수에서 timeoutInSeconds 및 timeoutPerCellInSeconds 필드를 설정하여 시간 제한을 변경할 수 있습니다.

Notebook 편집

이 메서드는 값이 있는 Notebook을 종료합니다. Notebook 또는 파이프라인에서 중첩 함수 호출을 실행할 수 있습니다.

Notebook에서 exit() 함수를 대화형으로 호출하면 Azure Synapse는 예외를 발생시키고 하위 시퀀스 셀 실행을 건너뛰며 Spark 세션을 활성 상태로 유지합니다.
exit() 함수를 호출하는 파이프라인에서 Notebook을 오케스트레이션하면 Notebook 활동이 종료 값으로 반환됩니다. 그러면 파이프라인 실행이 완료되고 Spark 세션이 중지됩니다.
참조되는 Notebook에서 exit() 함수를 호출하면 Fabric Spark는 참조된 Notebook의 추가 실행을 중지하고 run() 함수를 호출하는 주 Notebook에서 다음 셀을 계속 실행합니다. 예를 들어 Notebook1에는 세 개의 셀이 있으며 두 번째 셀에서 exit() 함수를 호출합니다. Notebook2에는 세 번째 셀에 5개의 셀과 호출 run(notebook1) 이 있습니다. Notebook2를 실행하면 exit() 함수를 누르면 Notebook1이 두 번째 셀에서 중지됩니다. Notebook2는 네 번째 셀과 다섯 번째 셀을 계속 실행합니다.

notebookutils.notebook.exit("value string")

참고 항목

exit() 함수는 현재 셀 출력을 덮어씁니다. 다른 코드 문의 출력이 손실되지 않도록 하려면 별도의 셀에서 notebookutils.notebook.exit()을 호출하세요.

예시:

다음 두 개의 셀이 있는 Sample1 Notebook:

셀 1은 기본값이 10으로 설정된 입력 매개 변수를 정의합니다.
셀 2는 입력을 종료 값으로 사용하여 Notebook을 종료합니다.

기본값을 사용하여 다른 Notebook에서 Sample1을 실행할 수 있습니다.

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

출력:

Notebook is executed successfully with exit value 10

다른 Notebook에서 Sample1을 실행하고 입력 값을 20으로 설정할 수 있습니다.

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

출력:

Notebook is executed successfully with exit value 20

Notebook 아티팩트 관리

notebookutils.notebook에서는 Notebook 항목을 프로그래밍 방식으로 관리하기 위한 특수 유틸리티를 제공합니다. 이러한 API는 Notebook 항목을 쉽게 만들고, 가져오고, 업데이트하고, 삭제하는 데 도움이 될 수 있습니다.

이러한 메서드를 효과적으로 활용하려면 다음 사용 예제를 고려하세요.

Notebook 만들기

with open("/path/to/notebook.ipynb", "r") as f:
    content = f.read()

artifact = notebookutils.notebook.create("artifact_name", "description", "content", "default_lakehouse_name", "default_lakehouse_workspace_id", "optional_workspace_id")

Notebook의 콘텐츠 가져오기

artifact = notebookutils.notebook.get("artifact_name", "optional_workspace_id")

Notebook 업데이트하기

updated_artifact = notebookutils.notebook.update("old_name", "new_name", "optional_description", "optional_workspace_id")

updated_artifact_definition = notebookutils.notebook.updateDefinition("artifact_name",  "content", "default_lakehouse_name", "default_Lakehouse_Workspace_name", "optional_workspace_id")

Notebook 삭제하기

is_deleted = notebookutils.notebook.delete("artifact_name", "optional_workspace_id")

작업 영역에서 Notebook 나열하기

artifacts_list = notebookutils.notebook.list("optional_workspace_id")

자격 증명 유틸리티

자격 증명 유틸리티를 사용하여 Azure Key Vault에서 액세스 토큰을 가져오고 비밀을 관리할 수 있습니다.

사용 가능한 메서드에 관한 개요를 가져오려면 다음 명령을 실행합니다.

notebookutils.credentials.help()

출력:

Help on module notebookutils.credentials in notebookutils:

NAME
    notebookutils.credentials - Utility for credentials operations in Fabric

FUNCTIONS
    getSecret(akvName, secret) -> str
        Gets a secret from the given Azure Key Vault.
        :param akvName: The name of the Azure Key Vault.
        :param secret: The name of the secret.
        :return: The secret value.
    
    getToken(audience) -> str
        Gets a token for the given audience.
        :param audience: The audience for the token.
        :return: The token.
    
    help(method_name=None)
        Provides help for the notebookutils.credentials module or the specified method.
        
        Examples:
        notebookutils.credentials.help()
        notebookutils.credentials.help("getToken")
        :param method_name: The name of the method to get help with.

DATA
    creds = <notebookutils.notebookutils.handlers.CredsHandler.CredsHandler...

FILE
    /home/trusted-service-user/cluster-env/trident_env/lib/python3.10/site-packages/notebookutils/credentials.py

토큰 가져오기

getToken은 지정된 대상 그룹 및 이름에 대한 Microsoft Entra 토큰을 반환합니다(선택 사항). 다음 목록에는 현재 사용 가능한 대상 그룹 키가 나와 있습니다.

스토리지 대상 리소스: "storage"
Power BI 리소스: "pbi"
Azure Key Vault 리소스: “keyvault”
Synapse RTA KQL DB 리소스: "kusto"

다음 명령을 실행하여 토큰을 가져옵니다.

notebookutils.credentials.getToken('audience Key')

사용자 자격 증명을 사용하여 비밀 얻기

getSecret은 사용자 자격 증명을 사용하여 지정된 Azure Key Vault 엔드포인트 및 비밀 이름에 대한 Azure Key Vault 비밀을 반환합니다.

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

파일 탑재 및 분리

Fabric은 Microsoft Spark 유틸리티 패키지에서 다음과 같은 탑재 시나리오를 지원합니다. 탑재, 분리, getMountPath()및 mounts() API를 사용하여 모든 작업 노드(드라이버 노드 및 작업자 노드)에 원격 스토리지(ADLS Gen2)를 연결할 수 있습니다. 스토리지 탑재 지점을 설정한 후에는 로컬 파일 API를 사용하여 로컬 파일 시스템에 저장된 것처럼 데이터에 액세스합니다.

ADLS Gen2 계정을 탑재하는 방법

다음 예제에서는 Azure Data Lake Storage Gen2를 탑재하는 방법을 보여 줍니다. Blob Storage 탑재 방법도 비슷합니다.

이 예제에서는 storegen2라는 Data Lake Storage Gen2 계정이 하나 있고 계정에 Notebook Spark 세션에 탑재/test하려는 mycontainer라는 컨테이너가 하나 있다고 가정합니다.

mycontainer라는 컨테이너를 탑재하려면 먼저 notebookutils에서 컨테이너에 액세스할 수 있는 권한이 있는지 확인해야 합니다. 현재 Fabric은 트리거 탑재 작업에 대해 accountKey 및 sastoken이라는 두 가지 인증 방법을 지원합니다.

공유 액세스 서명 토큰 또는 계정 키를 통해 탑재

NotebookUtils는 대상을 탑재하기 위한 매개 변수로 계정 키 또는 SAS(공유 액세스 서명) 토큰을 명시적으로 전달하도록 지원합니다.

보안상의 이유로 계정 키 또는 SAS 토큰을 Azure Key Vault에 저장하는 것이 좋습니다(다음 스크린샷에 표시됨). 그런 다음 notebookutils.credentials.getSecret API를 사용하여 검색할 수 있습니다. 자세한 내용은 Azure Key Vault 관리 스토리지 계정 키 정보를 참조하세요.

accountKey 메서드에 대한 샘플 코드:

# get access token for keyvault resource
# you can also use full audience here like https://vault.azure.net
accountKey = notebookutils.credentials.getSecret("<vaultURI>", "<secretName>")
notebookutils.fs.mount(  
    "abfss://mycontainer@<accountname>.dfs.core.windows.net",  
    "/test",  
    {"accountKey":accountKey}
)

sastoken에 대한 샘플 코드:

# get access token for keyvault resource
# you can also use full audience here like https://vault.azure.net
sasToken = notebookutils.credentials.getSecret("<vaultURI>", "<secretName>")
notebookutils.fs.mount(  
    "abfss://mycontainer@<accountname>.dfs.core.windows.net",  
    "/test",  
    {"sasToken":sasToken}
)

탑재 매개 변수:

fileCacheTimeout: Blob은 기본적으로 120초 동안 로컬 임시 폴더에 캐시됩니다. 이 시간 동안 blobfuse는 파일이 최신 상태인지 여부를 확인하지 않습니다. 매개 변수를 설정하여 기본 시간 제한 시간을 변경할 수 있습니다. 로컬 파일과 원격 파일 간의 불일치를 방지하기 위해 여러 클라이언트가 동시에 파일을 수정하는 경우 캐시 시간을 단축하거나 0으로 변경하고 항상 서버에서 최신 파일을 가져오는 것이 좋습니다.
제한 시간: 탑재 작업 제한 시간은 기본적으로 120초입니다. 매개 변수를 설정하여 기본 시간 제한 시간을 변경할 수 있습니다. 실행기가 너무 많거나 탑재 시간이 초과되는 경우 값을 늘리는 것이 좋습니다.

다음과 같은 매개 변수를 사용할 수 있습니다.

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

참고 항목

보안을 위해 코드에 자격 증명을 직접 포함하지 않는 것이 좋습니다. 자격 증명을 추가로 보호하기 위해 Notebook 출력에 표시되는 모든 비밀이 비공개 처리됩니다. 자세한 내용은 비밀 수정을 참조하세요.

Lakehouse를 탑재하는 방법

레이크하우스를 /<mount_name> 탑재하기 위한 샘플 코드:

notebookutils.fs.mount( 
 "abfss://<workspace_name>@onelake.dfs.fabric.microsoft.com/<lakehouse_name>.Lakehouse", 
 "/<mount_name>"
)

mssparktuils fs API를 사용하여 탑재 지점 아래의 파일에 액세스

탑재 작업의 주요 목적은 고객이 로컬 파일 시스템 API를 사용하여 원격 스토리지 계정에 저장된 데이터에 액세스할 수 있도록 하는 것입니다. 탑재된 경로가 있는 notebookutils fs API를 매개 변수로 사용하여 데이터에 액세스할 수도 있습니다. 이 경로 형식은 약간 다릅니다.

탑재 API를 사용하여 Data Lake Storage Gen2 컨테이너 mycontainer를 /test에 탑재했다고 가정합니다. 로컬 파일 시스템 API를 사용하여 데이터에 액세스할 때 경로 형식은 다음과 같습니다.

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

notebookutils fs API를 사용하여 데이터에 액세스하려는 경우 getMountPath()를 사용하여 정확한 경로를 가져오는 것이 좋습니다.

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

디렉터리 나열:

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

파일 콘텐츠 읽기:

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

디렉터리 만들기:

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

로컬 경로를 통해 탑재 지점 아래의 파일에 액세스

표준 파일 시스템을 사용하여 탑재 지점에서 파일을 쉽게 읽고 쓸 수 있습니다. 다음은 Python의 예입니다.

#File read
with open(notebookutils.fs.getMountPath('/test2') + "/myFile.txt", "r") as f:
    print(f.read())
#File write
with open(notebookutils.fs.getMountPath('/test2') + "/myFile.txt", "w") as f:
    print(f.write("dummy data"))

기존 탑재 지점을 확인하는 방법

Notebookutils.fs.mounts() API를 사용하여 모든 기존 탑재 지점 정보를 확인할 수 있습니다.

notebookutils.fs.mounts()

탑재 지점을 분리하는 방법

다음 코드를 사용하여 탑재 지점을 분리합니다(이 예제에서는 /test).

notebookutils.fs.unmount("/test")

알려진 제한 사항

현재 탑재는 작업 수준 구성입니다. 탑재 API를 사용하여 탑재 지점이 있는지 여부를 확인하는 것이 좋습니다.
분리 메커니즘은 자동으로 적용되지 않습니다. 애플리케이션 실행이 완료되면 탑재 지점을 분리하여 디스크 공간을 해제하려면 코드에서 분리 API를 명시적으로 호출해야 합니다. 그렇지 않으면 애플리케이션 실행이 완료된 후에도 탑재 지점이 노드에 계속 존재합니다.
ADLS Gen1 스토리지 계정 탑재는 지원되지 않습니다.

Lakehouse 유틸리티

notebookutils.lakehouse는 Lakehouse 항목을 관리하기 위한 맞춤형 유틸리티를 제공합니다. 이러한 유틸리티를 사용하면 Lakehouse 아티팩트 만들기, 가져오기, 업데이트 및 삭제를 손쉽게 수행할 수 있습니다.

방법 개요

다음은 notebookutils.lakehouse제공하는 사용 가능한 방법에 대한 개요입니다.

# Create a new Lakehouse artifact
create(name: String, description: String = "", definition: ItemDefinition = null, workspaceId: String = ""): Artifact

# Retrieve a Lakehouse artifact
get(name: String, workspaceId: String = ""): Artifact

# Get a Lakehouse artifact with properties
getWithProperties(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 = "", maxResults: Int = 1000): Array[Artifact]

# List all tables in a Lakehouse artifact
listTables(lakehouse: String, workspaceId: String = "", maxResults: Int = 1000): Array[Table] 

# Starts a load table operation in a Lakehouse artifact
loadTable(loadOption: collection.Map[String, Any], table: String, lakehouse: String, workspaceId: String = ""): Array[Table]

사용 예

이러한 메서드를 효과적으로 활용하려면 다음 사용 예제를 고려하세요.

Lakehouse 만들기

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

Lakehouse 가져오기

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

artifact = notebookutils.lakehouse.getWithProperties("artifact_name", "optional_workspace_id")

Lakehouse 업데이트하기

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

Lakehouse 삭제하기

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

작업 영역에 Lakehouses 나열하기

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

Lakehouse의 모든 테이블 나열하기

artifacts_tables_list = notebookutils.lakehouse.listTables("artifact_name", "optional_workspace_id")

Lakehouse에서 테이블 로드 작업 시작하기

notebookutils.lakehouse.loadTable(
    {
        "relativePath": "Files/myFile.csv",
        "pathType": "File",
        "mode": "Overwrite",
        "recursive": False,
        "formatOptions": {
            "format": "Csv",
            "header": True,
            "delimiter": ","
        }
    }, "table_name", "artifact_name", "optional_workspace_id")

추가 정보

각 방법과 해당 매개변수에 대한 자세한 정보를 확인하려면 notebookutils.lakehouse.help("methodName") 함수를 활용하세요.

런타임 유틸리티

세션 컨텍스트 정보 표시

notebookutils.runtime.context를 사용하면 Notebook 이름, 기본 lakehouse, 워크스페이스 정보, 파이프라인 실행 여부 등 현재 라이브 세션의 컨텍스트 정보를 얻을 수 있습니다.

notebookutils.runtime.context

세션 관리

대화형 세션 중지

중지 단추를 수동으로 클릭하는 대신 코드에서 API를 호출하여 대화형 세션을 중지하는 것이 더 편리한 경우도 있습니다. 이러한 경우 코드를 통해 대화형 세션 중지를 지원하는 API notebookutils.session.stop() 제공하며 Scala 및 PySpark에서 사용할 수 있습니다.

notebookutils.session.stop()

notebookutils.session.stop() API는 백그라운드에서 현재 대화형 세션을 비동기적으로 중지합니다. 또한 Spark 세션을 중지하고 세션이 차지하는 리소스를 해제하므로 동일한 풀의 다른 세션에서 사용할 수 있습니다.

Python 인터프리터 다시 시작

notebookutils.session 유틸리티는 Python 인터프리터를 다시 시작하는 방법을 제공합니다.

notebookutils.session.restartPython()

참고 항목

Notebook 참조 실행의 경우, restartPython()은 참조되는 현재 Notebook의 Python 인터프리터만 다시 시작합니다.
드물게 Spark 리플렉션 메커니즘으로 인해 명령이 실패할 수 있으므로 재시도를 추가하면 문제가 완화됩니다.

알려진 문제

1.2 이상의 런타임 버전을 사용하고 notebookutils.help()를 실행하는 경우, 나열된 fabricClient, PBIClient API는 현재 지원되지 않으며 추후 제공될 예정입니다. 또한 자격 증명 API는 현재 Scala Notebook에서 지원되지 않습니다.
Python Notebook은 세션 관리를 위해 notebookutils.session 유틸리티를 사용할 때 정지,, , APIs를 지원하지 않습니다.

다음을 통해 공유

Fabric용 NotebookUtils(이Notebook MSSparkUtils)

파일 시스템 유틸리티

파일 나열

파일 속성 보기

새 디렉터리 만들기

파일 복사

성능 복사 파일

파일 콘텐츠 미리 보기

파일 이동

파일 쓰기

파일에 콘텐츠 추가

파일 또는 디렉터리 삭제

탑재/분리 디렉터리

Notebook 유틸리티

Notebook 참조

참조는 여러 Notebook을 병렬로 실행합니다.

Notebook 편집

Notebook 아티팩트 관리

Notebook 만들기

Notebook의 콘텐츠 가져오기

Notebook 업데이트하기

Notebook 삭제하기

작업 영역에서 Notebook 나열하기

자격 증명 유틸리티

토큰 가져오기

사용자 자격 증명을 사용하여 비밀 얻기

파일 탑재 및 분리

ADLS Gen2 계정을 탑재하는 방법

공유 액세스 서명 토큰 또는 계정 키를 통해 탑재

Lakehouse를 탑재하는 방법

mssparktuils fs API를 사용하여 탑재 지점 아래의 파일에 액세스

로컬 경로를 통해 탑재 지점 아래의 파일에 액세스

기존 탑재 지점을 확인하는 방법

탑재 지점을 분리하는 방법

알려진 제한 사항

Lakehouse 유틸리티

방법 개요

사용 예

Lakehouse 만들기

Lakehouse 가져오기

Lakehouse 업데이트하기

Lakehouse 삭제하기

작업 영역에 Lakehouses 나열하기

Lakehouse의 모든 테이블 나열하기

Lakehouse에서 테이블 로드 작업 시작하기

추가 정보

런타임 유틸리티

세션 컨텍스트 정보 표시

세션 관리

대화형 세션 중지

Python 인터프리터 다시 시작

알려진 문제

관련 콘텐츠

피드백

추가 리소스