Databricks PySpark中master_dim.py能否并行调用dim_1.py、dim_2.py运行

可行性说明 #

Databricks PySpark完全支持通过master脚本并行调用多个子脚本执行，你描述的调度场景无需额外开通付费能力即可实现。

必要配置与正确实现步骤 #

• 禁用Python原生multiprocessing做调度：Databricks执行环境默认限制进程fork权限，使用原生多进程会直接触发权限报错或子脚本无响应，优先使用concurrent.futures.ThreadPoolExecutor做线程级并行调度，若你用的是Databricks notebook形态的脚本，也可以用dbutils.notebook.run的异步调用模式。
• 提前添加脚本路径到系统路径：master脚本执行前必须将子脚本所在的目录添加到Python sys.path中，否则会出现模块导入失败的报错，示例配置：

import sys
# 路径替换为dim_1.py、dim_2.py实际存放的Workspace路径
sys.path.append("/Workspace/Users/xxx@xxx.com/dim_scripts/")

• 开启Spark会话共享配置：若两个子脚本均包含Spark计算逻辑，需要提前在集群的Spark配置项中添加spark.databricks.session.share.enabled = true，避免多个子脚本的Spark会话冲突报错，也可以在子脚本开头给Spark作业设置独立分组，方便后续日志排查：

# dim_1.py/dim_2.py开头添加
spark.sparkContext.setJobGroup("dim_1_calc", "维度1数据加工作业")

• 正确的master_dim.py执行逻辑示例：

from concurrent.futures import ThreadPoolExecutor, as_completed
import sys
sys.path.append("/Workspace/Users/xxx@xxx.com/dim_scripts/")
# 子脚本需提前封装统一的run入口函数，所有执行逻辑封装在run方法内
import dim_1
import dim_2

def execute_dim_script(script_tag: str):
    if script_tag == "dim1":
        return dim_1.run()
    elif script_tag == "dim2":
        return dim_2.run()

# 并行度根据集群可用资源设置，此处两个脚本设为2即可
with ThreadPoolExecutor(max_workers=2) as pool:
    tasks = [pool.submit(execute_dim_script, tag) for tag in ["dim1", "dim2"]]
    for task in as_completed(tasks):
        try:
            task.result()
        except Exception as e:
            print(f"子脚本执行异常：{str(e)}")

常见报错排查方向 #

• 若出现模块导入错误：优先检查sys.path添加的路径是否正确，确认子脚本无语法错误
• 若出现Spark权限/会话冲突错误：检查集群是否开启了会话共享配置，确认两个子脚本没有修改全局Spark配置的逻辑
• 若出现子脚本执行卡住：检查集群资源是否足够同时运行两个子脚本的Spark作业，可适当调大集群内核数或者降低并行度

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