Azure Synapse中PySpark如何创建多笔记本共享Spark会话

核心根因说明 #

• 你观察到的appName显示为平台默认值、手动停止会话后重建报错，本质是Azure Synapse托管Notebook的固定运行机制：所有Synapse笔记本在启动时，平台会自动注入全局默认的SparkContext，不允许用户手动终止这个托管上下文，也不支持覆盖平台默认的appName等基础配置。手动调用sc.stop()会直接打断和YARN ApplicationMaster绑定的应用生命周期，这就是你遇到Promise already completed报错的直接原因。
• 之前每处理一个文件就触发3分钟初始化开销的根源，不是你代码里的getOrCreate()逻辑有问题，是ADF管道ForEach循环内部的每个独立Notebook活动，默认都会触发一次Spark池冷启动：哪怕两个Notebook指向同一个Spark池，只要是独立的活动节点，就会重新申请计算资源、完成会话初始化，这部分开销根本省不掉。

不需要纠结appName的显示值：Synapse平台会强制覆盖用户自定义的appName配置，你写的.appName("837App")代码从执行开始就不会生效，看到平台默认的appName是完全正常的现象，不代表你的会话没有复用。

正确的共享会话实现路径 #

1. 重构流程逻辑，消除循环内的独立Notebook活动 #

你之前的架构是在ADF ForEach循环容器里，给每个待处理文件单独配置一个Notebook活动 —— 这是冷启动开销的最大来源。正确的复用方式是：

• 把整个文件遍历、数据转换、写表的逻辑全部移到单个Notebook内部完成，删除ADF层面的ForEach循环容器。Notebook内部自己完成ADLS Gen2的文件枚举、逐文件处理，所有逻辑跑在同一次Spark会话里，完全不会触发循环节点的冷启动。
• 调整存储过程活动的位置：把sproc执行从循环内部移到所有文件处理完成之后再触发，减少不必要的节点跳转开销。

参考实现代码：

from pyspark.sql import SparkSession
import re
from pyspark.sql.functions import row_number, monotonically_increasing_id
from pyspark.sql.window import Window

# 直接使用平台自动注入的全局spark对象，不需要手动新建会话
# 仅需调整空闲超时配置，避免处理过程中会话被提前回收
spark.conf.set("spark.synapse.pool.executor.idleTimeout", "3600")

# 1. 从ADLS Gen2枚举所有待处理的837格式CSV文件
file_list = mssparkutils.fs.ls(f"{srcPath}/{srcFilesDirectory}/")
target_files = [f for f in file_list if f.name.endswith('.csv') and ('837i' in f.name or '837p' in f.name)]

# 2. 逐文件处理，全程复用同一个Spark会话
for file in target_files:
    srcFQFN = file.path
    dstTableName = "raw_837i" if "837i" in file.name else "raw_837p"
    
    # 原有数据清洗逻辑保持不变
    df = spark.read.load(srcFQFN, format='csv', delimiter=f"{srcFileDelimeter}", header=True)
    adf = df.withColumn('AutoID', row_number().over(Window.orderBy(monotonically_increasing_id())))
    adf = adf.toDF(*(re.sub(r"[\.\s\(\)\-]+", "_", c) for c in adf.columns))
    
    spark.sql(f"CREATE DATABASE IF NOT EXISTS {sparkDbName}")
    adf.write.mode("append").saveAsTable(f"{sparkDbName}.{dstTableName}")

# 所有文件处理完成后，统一触发存储过程逻辑
mssparkutils.notebook.run("call_sproc_notebook", 600, {"targetDb": sparkDbName})

按这个逻辑重构后，你之前3个文件、共22行数据的处理流程，只会产生1次3分钟的会话初始化开销，实际数据处理环节仅需数秒，整体运行时长会从12分钟降到4分钟左右，后续新增更多文件也不会再叠加会话初始化成本。

2. 会话复用状态的验证方法 #

因为平台会强制覆盖自定义appName，不能靠返回的appName值判断会话是否复用，可以通过两个可靠方式验证：

• 打印当前会话的全局唯一标识：

# 输出当前Spark会话的YARN应用ID
print(spark.sparkContext.applicationId)

只要所有处理步骤输出的applicationId完全一致，就说明是同一个会话在执行，不需要关注显示的appName值。

• 在Synapse Studio的监视面板中查看Spark应用列表，确认整个流程只触发了1个YARN应用，就代表没有重复初始化会话。

3. 常见操作避坑 #

• 永远不要在Synapse Notebook中主动调用sc.stop()：这个操作会直接打断平台托管的YARN应用生命周期，触发你之前遇到的Promise already completed报错。
• 你之前额外加的“初始化Notebook”环节是多余的：Synapse Notebook启动时已经自动完成了SparkContext的初始化，不需要手动调用SparkSession.builder.getOrCreate()新建会话，直接使用全局的spark对象即可，这个对象本身就是平台初始化好的共享实例。
• 不要指望靠同一个Spark池实现跨ADF活动的会话复用：哪怕两个Notebook活动指向同一个Spark池，只要是独立的ADF活动节点，就会启动新的YARN应用，冷启动开销无法避免。

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