通过

使用 NotebookUtils 管理笔记本工件

在 Microsoft Fabric 中使用 notebookutils.notebook 以编程方式管理笔记本项目。 可以创建、检索、更新、删除和列出笔记本项目，以自动执行部署、生命周期管理和 CI/CD 工作流。

注释

这些 API 仅在 Fabric 笔记本中受支持，而不支持在 Azure Synapse 中。 对于每个操作，必须在目标工作区中具有适当的权限。

下表列出了可用的笔记本管理方法：

方法 Signature 说明
create create(name, description, content, defaultLakehouse, defaultLakehouseWorkspace, workspaceId): Artifact 创建新的笔记本。
get get(name, workspaceId): Artifact 按名称或 ID 检索笔记本。
getDefinition getDefinition(name, workspaceId, format): String 检索笔记本定义（内容）。
update update(name, newName, description, workspaceId): Artifact 更新笔记本元数据。
updateDefinition updateDefinition(name, content, defaultLakehouse, defaultLakehouseWorkspace, workspaceId, environmentId, environmentWorkspaceId): bool 更新笔记本定义和湖仓。
delete delete(name, workspaceId): Boolean 删除笔记本。
list list(workspaceId, maxResults): Array[Artifact] 列出工作区中的所有笔记本。

创建笔记本

使用 notebookutils.notebook.create() 在当前工作区或指定工作区中创建新的笔记本工件。

注释

本文中读取或写入 .ipynb 文件的工作流示例将 Python 用于文件 I/O。 除非另有说明，否则核心 notebookutils.notebook API 在 Python、PySpark、Scala 和 R 中可用。

参数

参数 类型 必需 说明
name String 是的 新笔记本的显示名称。 在工作区中必须是独一无二的。
description String 否 笔记本的说明。 默认值为空。
content 字符串、字节或字典 是的 采用有效 .ipynb JSON 格式的笔记本内容。 也可以是原始字节或字典对象。 不能为空。
defaultLakehouse String 否 要附加的默认 Lakehouse 的名称或 ID。
defaultLakehouseWorkspace String 否 默认 Lakehouse 的工作区 ID。 为当前工作区留空。
workspaceId String 否 目标工作区 ID。 为当前工作区留空。

重要

参数 content 不能为空。 创建笔记本时，必须提供有效的 .ipynb 格式内容。 至少提供有效的空笔记本结构：

{
  "cells": [],
  "metadata": {},
  "nbformat": 4,
  "nbformat_minor": 5
}

从模板创建笔记本

# Read notebook template from a file
with open("/path/to/template.ipynb", "r") as f:
    notebook_content = f.read()

# Create the notebook
notebook = notebookutils.notebook.create(
    name="ProcessingNotebook",
    description="Data processing notebook from template",
    content=notebook_content
)

print(f"Created notebook: {notebook.displayName} (ID: {notebook.id})")

使用默认 Lakehouse 创建笔记本

# Minimum valid notebook content - content cannot be empty
minimal_content = '''{
    "cells": [],
    "metadata": {},
    "nbformat": 4,
    "nbformat_minor": 5
}'''

# Create notebook with default lakehouse configuration
notebook = notebookutils.notebook.create(
    name="DataAnalysis",
    description="Analysis notebook with lakehouse access",
    content=minimal_content,
    defaultLakehouse="MyLakehouse",
    defaultLakehouseWorkspace=""  # Current workspace
)

print(f"Created notebook with lakehouse: {notebook.displayName}")

返回值

该方法 create() 返回 Artifact 具有以下属性的对象：

  • displayName：笔记本的显示名称。
  • id：已创建的笔记本的唯一标识符。
  • description：笔记本的说明。

在另一个工作区中创建笔记本

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

notebook = notebookutils.notebook.create(
    name="SharedNotebook",
    description="Notebook for the shared workspace",
    content=content,
    workspaceId="bbbbbbbb-2222-3333-4444-cccccccccccc"
)

print(f"Created in remote workspace: {notebook.displayName}")

从模板创建多个笔记本

# Load template content (must be valid .ipynb)
with open("/path/to/template.ipynb", "r") as f:
    template_content = f.read()

regions = ["US", "EU", "Asia"]

created_notebooks = []
for region in regions:
    notebook = notebookutils.notebook.create(
        name=f"Process_{region}",
        description=f"Processing notebook for {region} region",
        content=template_content,
        defaultLakehouse=f"Lakehouse_{region}"
    )
    created_notebooks.append(notebook)
    print(f"Created: {notebook.displayName}")

print(f"\nCreated {len(created_notebooks)} notebooks")

小窍门

为笔记本提供有意义的名称和说明，使其更易于查找。 使用一致的命名约定，例如在自动部署时使用<Project>_<Purpose>_<Region>。

获取笔记本

使用notebookutils.notebook.get()按名称或 ID 检索笔记本元数据。 它返回一个 Artifact 具有属性的对象，例如 displayName， id和 description。

参数

参数 类型 必需 说明
name String 是的 要检索的笔记本的名称或 ID。
workspaceId String 否 工作区 ID。 为当前工作区留空。

从当前工作区获取笔记本

notebook = notebookutils.notebook.get("MyNotebook")

print(f"Notebook Name: {notebook.displayName}")
print(f"Notebook ID: {notebook.id}")
print(f"Description: {notebook.description}")

从另一个工作区获取笔记本

workspace_id = "bbbbbbbb-2222-3333-4444-cccccccccccc"
notebook = notebookutils.notebook.get("SharedNotebook", workspaceId=workspace_id)

print(f"Retrieved: {notebook.displayName} from workspace {workspace_id}")

返回值

该方法 get() 返回 Artifact 具有以下属性的对象：

  • displayName：笔记本的显示名称。
  • id：唯一标识符。
  • description：笔记本的说明。

小窍门

在更新或删除操作之前使用 get() ，验证目标笔记本是否存在。 还可以使用它来检查笔记本名称是否已在使用中，然后再创建新名称。

获取笔记本定义

使用 notebookutils.notebook.getDefinition() 格式检索完整的笔记本内容 .ipynb。 使用它进行备份、迁移、版本控制或内容分析。

参数

参数 类型 必需 说明
name String 是的 笔记本的名称或 ID。
workspaceId String 否 工作区 ID。 为当前工作区留空。
format String 否 输出格式。 默认为 "ipynb"。

检索和保存笔记本定义

# Retrieve notebook definition as .ipynb content
notebook_content = notebookutils.notebook.getDefinition("MyNotebook")

# Save to a file for backup
with open("/path/to/backup/MyNotebook.ipynb", "w") as f:
    f.write(notebook_content)

print("Notebook definition retrieved and saved")

从另一个工作区获取笔记本定义

workspace_id = "cccccccc-3333-4444-5555-dddddddddddd"
notebook_content = notebookutils.notebook.getDefinition(
    name="SharedNotebook",
    workspaceId=workspace_id,
    format="ipynb"
)

print(f"Retrieved definition from workspace {workspace_id}")

返回值

该方法 getDefinition() 返回一个字符串，其中包含 JSON 格式的 .ipynb 笔记本内容。

导出所有笔记本进行备份

import os
from datetime import datetime

def export_all_notebooks(backup_dir="/path/to/backups"):
    """Export all notebooks in the workspace for backup."""

    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    export_dir = f"{backup_dir}/backup_{timestamp}"
    os.makedirs(export_dir, exist_ok=True)

    notebooks = notebookutils.notebook.list()
    print(f"Exporting {len(notebooks)} notebooks to {export_dir}")

    exported_count = 0
    for nb in notebooks:
        try:
            content = notebookutils.notebook.getDefinition(nb.displayName)
            filename = f"{export_dir}/{nb.displayName}.ipynb"
            with open(filename, "w") as f:
                f.write(content)
            exported_count += 1
            print(f"Exported: {nb.displayName}")
        except Exception as e:
            print(f"Failed to export {nb.displayName}: {e}")

    print(f"\nExported {exported_count} of {len(notebooks)} notebooks")
    return export_dir

backup_location = export_all_notebooks()

更新笔记本

使用 notebookutils.notebook.update() 更改笔记本元数据，例如其显示名称和说明。 它不会修改笔记本内容或 Lakehouse 配置。

参数

参数 类型 必需 说明
name String 是的 笔记本的当前名称或 ID。
newName String 是的 笔记本的新显示名称。
description String 否 更新了说明。
workspaceId String 否 工作区 ID。 请为当前工作区留空。

重命名笔记本

updated_notebook = notebookutils.notebook.update(
    name="OldNotebookName",
    newName="NewNotebookName",
    description="Updated description with more details"
)

print(f"Updated notebook: {updated_notebook.displayName}")

返回值

该方法 update() 返回一个 Artifact 具有更新属性的对象。

更新笔记本定义

使用 notebookutils.notebook.updateDefinition() 修改笔记本内容、默认湖仓或两者同时修改。 如果需要更改笔记本定义而不是其元数据，请使用它。

参数

参数 类型 必需 说明
name String 是的 要更新的笔记本的名称或 ID。
content String 否 新的笔记本内容采用 .ipynb 格式。
defaultLakehouse String 否 新的默认 Lakehouse 名称。
defaultLakehouseWorkspace String 否 新默认 Lakehouse 的工作区ID。 为当前工作区留空。
workspaceId String 否 工作区 ID。 为当前工作区留空。
environmentId String 否 要附加到笔记本的环境 ID。
environmentWorkspaceId String 否 环境的工作区 ID。 请为当前工作区留空。

注释

和environmentIdenvironmentWorkspaceId参数仅在 Spark 笔记本运行时中可用。 Python 笔记本不支持这些参数。

更新笔记本内容

# Load new content
with open("/path/to/updated_notebook.ipynb", "r") as f:
    new_content = f.read()

is_updated = notebookutils.notebook.updateDefinition(
    name="MyNotebook",
    content=new_content
)

print(f"Notebook definition updated: {is_updated}")

更改默认 Lakehouse

is_updated = notebookutils.notebook.updateDefinition(
    name="MyNotebook",
    defaultLakehouse="NewLakehouse",
    defaultLakehouseWorkspace=""  # Current workspace
)

print(f"Default lakehouse updated: {is_updated}")

更新内容和 Lakehouse

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

is_updated = notebookutils.notebook.updateDefinition(
    name="MyNotebook",
    content=new_content,
    defaultLakehouse="ProductionLakehouse",
    defaultLakehouseWorkspace=""
)

print(f"Notebook fully updated: {is_updated}")

返回值

此方法 updateDefinition() 返回 True 更新成功或 False 失败。

小窍门

使用 update() 进行元数据更改（如名称、说明）以及 updateDefinition() 进行内容和 Lakehouse 的更改。 需要对元数据和内容进行完全刷新时，按顺序调用这两种方法。

删除笔记本

使用 notebookutils.notebook.delete() 从工作区中永久删除笔记本。 如果删除成功，返回True；否则，返回False。

参数

参数 类型 必需 说明
name String 是的 要删除的笔记本的名称或 ID。
workspaceId String 否 工作区 ID。 为当前工作区留空。

重要

删除是一项永久性操作。 无法恢复已删除的笔记本。 在删除之前，请务必核实笔记本名称，并考虑先 getDefinition() 备份笔记本的定义。

返回值

此方法 delete() 返回 True 删除成功或 False 失败的情况。

删除笔记本

is_deleted = notebookutils.notebook.delete("ObsoleteNotebook")

if is_deleted:
    print("Notebook deleted successfully")
else:
    print("Failed to delete notebook")

按模式安全地清理笔记本

def cleanup_notebooks(name_pattern, dry_run=True):
    """Delete notebooks matching a name pattern."""

    notebooks = notebookutils.notebook.list()
    to_delete = [nb for nb in notebooks if name_pattern in nb.displayName]

    print(f"Found {len(to_delete)} notebooks matching '{name_pattern}':")
    for nb in to_delete:
        print(f"  - {nb.displayName}")

    if dry_run:
        print("\nDRY RUN - No notebooks deleted")
        return

    deleted_count = 0
    for nb in to_delete:
        if notebookutils.notebook.delete(nb.displayName):
            deleted_count += 1
            print(f"Deleted: {nb.displayName}")
        else:
            print(f"Failed to delete: {nb.displayName}")

    print(f"\nDeleted {deleted_count} of {len(to_delete)} notebooks")

# Always run with dry_run=True first to preview
cleanup_notebooks("temp_", dry_run=True)

小窍门

若要安全批量删除，请始终先运行 dry_run=True ，以预览将删除哪些笔记本。 请考虑使用 _TO_DELETE 前缀重命名笔记本，而不是立即删除它们，以便根据需要恢复它们。

列出笔记本

使用 notebookutils.notebook.list() 枚举工作区中的笔记本。 它返回对象数组 Artifact 。

参数

参数 类型 必需 说明
workspaceId String 否 工作区 ID。 为当前工作区留空。
maxResults int (整数) 否 要返回的最大结果数。 默认值为1000。

列出当前工作区中的所有笔记本

notebooks = notebookutils.notebook.list()

print(f"Found {len(notebooks)} notebooks:")
for nb in notebooks:
    print(f"  - {nb.displayName} (ID: {nb.id})")

列出另一个工作区中的笔记本

workspace_id = "cccccccc-3333-4444-5555-dddddddddddd"
notebooks = notebookutils.notebook.list(workspaceId=workspace_id)

print(f"Found {len(notebooks)} notebooks in workspace {workspace_id}")

返回值

该方法 list() 返回对象数组 Artifact 。 每个对象都包含 displayName、 id属性和 description 属性。

按名称模式筛选笔记本

all_notebooks = notebookutils.notebook.list()

# Filter for notebooks that start with a specific prefix
processing_notebooks = [nb for nb in all_notebooks if nb.displayName.startswith("Process_")]

print(f"Found {len(processing_notebooks)} processing notebooks:")
for nb in processing_notebooks:
    print(f"  - {nb.displayName}")

克隆笔记本

在同一工作区内或其他工作区中，使用 list() 和 getDefinition() 一起克隆笔记本。

def clone_notebook(source_name, target_name, target_workspace=""):
    """Clone a notebook by retrieving its content and creating a copy."""

    source = notebookutils.notebook.get(source_name)
    content = notebookutils.notebook.getDefinition(source_name)

    cloned = notebookutils.notebook.create(
        name=target_name,
        description=f"Clone of {source_name}",
        content=content,
        workspaceId=target_workspace
    )

    print(f"Cloned {source_name} to {cloned.displayName}")
    return cloned

cloned_notebook = clone_notebook("TemplateNotebook", "NewInstance")

将笔记本迁移到另一个工作区

def migrate_notebook(name, target_workspace_id, new_name=None):
    """Migrate a notebook from the current workspace to another workspace."""

    content = notebookutils.notebook.getDefinition(name)
    target_name = new_name if new_name else name

    migrated = notebookutils.notebook.create(
        name=target_name,
        description=f"Migrated from {name}",
        content=content,
        workspaceId=target_workspace_id
    )

    print(f"Migrated {name} to workspace {target_workspace_id} as {target_name}")
    return migrated

target_ws = "dddddddd-4444-5555-6666-eeeeeeeeeeee"
migrated_nb = migrate_notebook("DataPipeline", target_ws, "DataPipeline_v2")

相关内容

  • Last updated on