调用mssparkutils.notebook.run()后如何调用其他Synapse笔记本定义的函数

mssparkutils.notebook.run() 会在独立的隔离Spark会话中运行目标笔记本，执行结束后该会话会被直接回收，目标笔记本内定义的函数、变量、临时对象都不会同步到当前调用笔记本的会话上下文，因此直接调用函数会抛出NameError: name 'load_cosmos_data' is not defined。
%run魔法命令的逻辑是将目标笔记本的全量代码拉取到当前会话顺序执行，因此可以直接共享函数和变量，但原生%run的参数传递仅支持硬编码字面量，无法直接传入动态变量。

方案1：将函数封装为独立Python模块（生产环境优先推荐） #

把存储在笔记本中的函数逻辑迁移到工作区存储的.py脚本文件中，通过Python原生import逻辑导入使用，完全绕开会话隔离问题，参数传递不受限制，也方便版本管理和多项目复用。

• 在Synapse工作区新建工具目录（例如/Shared/Utils/），将原"function definitions"笔记本中的函数代码存为目录下的cosmos_utils.py文件，示例内容：

# cosmos_utils.py
def load_cosmos_data(param_val):
    # 替换为原有的Cosmos数据加载逻辑，直接使用传入的param_val参数
    return spark.read.format("cosmos.oltp")\
        .option("spark.synapse.linkedService", "你的Cosmos链接服务名")\
        .option("spark.cosmos.container", "目标容器名")\
        .load()\
        .filter(f"category = '{param_val}'")

• 在需要调用函数的笔记本中，先将工具脚本所在目录加入Python搜索路径，再导入函数传参调用即可：

import sys
# 替换为实际存放工具脚本的工作区绝对路径
sys.path.append("/synapse/workspaces/你的工作区名称/Shared/Utils/")

# 导入目标函数
from cosmos_utils import load_cosmos_data

# 定义动态参数
value = "test"
# 直接传参调用
df = load_cosmos_data(param_val=value)

方案2：通过notebook返回值+exec动态注册函数（仅适合快速调试场景） #

如果不想改动现有笔记本结构，可以调整目标笔记本的返回逻辑，将函数源码和参数初始化代码作为返回值输出，在调用方拿到代码后通过exec()在当前会话动态注册函数。

• 修改存储函数的"function definitions"笔记本，在末尾添加返回逻辑，将函数源码和参数拼接为可执行的Python代码后退出：

# "function definitions"笔记本原有逻辑
# 参数单元格
param = "default"
# 原有函数定义
def load_cosmos_data():
    # 替换为原有的Cosmos数据加载逻辑，直接使用上下文中的param变量
    return spark.read.format("cosmos.oltp")\
        .option("spark.synapse.linkedService", "你的Cosmos链接服务名")\
        .option("spark.cosmos.container", "目标容器名")\
        .load()\
        .filter(f"category = '{param}'")

# 新增返回逻辑：拼接可执行代码块
import inspect
func_source = inspect.getsource(load_cosmos_data)
exec_code = f"""
param = "{param}"
{func_source}
"""
# 退出笔记本并返回代码字符串
mssparkutils.notebook.exit(exec_code)

• 在调用方笔记本中运行目标笔记本，拿到返回代码后执行注册：

from notebookutils import mssparkutils

value = "test"
# 运行目标笔记本，传入动态参数
exec_code = mssparkutils.notebook.run("function definitions", 60, {"param": value})
# 在当前会话执行返回的代码，完成函数注册
exec(exec_code)
# 直接调用函数即可
df = load_cosmos_data()

注意：该方案需要确保函数依赖的所有变量、导入包都被包含在返回的代码块中，否则执行时会抛出变量未定义错误；同时该方案存在代码注入风险，禁止在生产环境使用。

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