Databricks中能否通过其他笔记本批量更新笔记本的%run路径？

完全可以通过单个Python笔记本完成全量路径替换，不需要接入第三方工具，核心是调用Databricks内置的Workspace接口遍历、修改笔记本内容，操作步骤如下：

前置注意 #

• 执行操作的账号需要对所有待修改笔记本持有读写权限，避免操作中途权限报错
• 正式修改前建议先把待扫描的笔记本目录整体导出备份，方便误操作时回滚
• 首次运行建议开启预览模式，只输出待修改的笔记本列表，确认匹配范围无误后再执行实际写入

实现代码 #

直接在新建的Databricks Python笔记本单元格中粘贴以下代码，修改开头的配置项即可运行：

import base64
from pyspark.dbutils import DBUtils

dbutils = DBUtils(spark)

# -------------------------- 按需修改以下配置 --------------------------
SCAN_ROOT_PATH = "/Shared/your_target_notebook_dir"  # 填写需要扫描修改的笔记本所在根目录
OLD_RUN_STATEMENT = '%run "/Shared/folder/script/notebook"'  # 待替换的旧%run语句
NEW_RUN_STATEMENT = '%run "/Repos/Dev/folder/script/notebook"'  # 替换后的新%run语句
DRY_RUN = True  # 预览模式开关：True=仅打印变更不写入，False=实际执行替换
# --------------------------------------------------------------------

def get_all_notebooks(path: str) -> list:
    """递归遍历目录下所有笔记本文件"""
    file_list = dbutils.fs.ls(path)
    notebooks = []
    for item in file_list:
        if item.isDir():
            notebooks.extend(get_all_notebooks(item.path))
        else:
            # 过滤非笔记本文件：Databricks笔记本路径无文件后缀
            file_name = item.path.rstrip("/").split("/")[-1]
            if "." not in file_name:
                notebooks.append(item.path.rstrip("/"))
    return notebooks

def update_single_notebook(nb_path: str):
    """处理单个笔记本的内容替换"""
    # 调用Workspace API导出笔记本源码
    api_client = dbutils.notebook.entry_point.getDbutils().notebook().getContext().apiClient()
    export_res = api_client.do(
        "GET",
        "/api/2.0/workspace/export",
        {"path": nb_path, "format": "SOURCE"}
    )
    origin_content = base64.b64decode(export_res["content"]).decode("utf-8")
    updated_content = origin_content.replace(OLD_RUN_STATEMENT, NEW_RUN_STATEMENT)
    
    if origin_content == updated_content:
        print(f"[跳过] {nb_path} 无匹配的旧路径，无需修改")
        return
    
    print(f"[命中] {nb_path} 存在待替换路径")
    if DRY_RUN:
        return
    
    # 回写修改后的内容
    encoded_content = base64.b64encode(updated_content.encode("utf-8")).decode("utf-8")
    api_client.do(
        "POST",
        "/api/2.0/workspace/import",
        {
            "path": nb_path,
            "format": "SOURCE",
            "content": encoded_content,
            "overwrite": True
        }
    )
    print(f"[完成] {nb_path} 路径替换成功")

# 执行批量处理
all_nbs = get_all_notebooks(SCAN_ROOT_PATH)
print(f"扫描完成，共找到{len(all_nbs)}个笔记本，开始检查匹配：")
for nb in all_nbs:
    update_single_notebook(nb)

补充说明 #

• 代码无额外依赖，不需要安装第三方包，Databricks全版本运行时都支持
• 该逻辑对Python、SQL、Scala、R等所有语言的笔记本都生效，%run是Databricks全局magic命令，所有笔记本都以纯文本格式存储源码
• 如果你的旧%run语句后面附带了参数传递（比如$env这类变量），只需要调整OLD_RUN_STATEMENT的匹配内容即可，替换逻辑不需要改动
• 替换完成后建议抽查2-3个笔记本确认路径正确，再运行关联任务验证Repos路径下的脚本可以正常调用

内容的提问来源于stack exchange，提问作者Rchee