You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

PySpark调用Pandas on Spark API获取DataFrame shape报模块导入错误

avatar
算力阿峰
2026-8-30
问题背景

我的项目当前目录结构如下：

|- logger
|--- __init__.py
|--- logger.py
|- another_package
|--- __init__.py
|--- module1.py
|- models
|--- model1
|------ main.py
|------ model1_utilities.py

Spark Context与Spark Session均在main.py中完成初始化启动，main.py调用model1_utilities.py的核心逻辑如下：

results = function_a(params)
logger.info('Calculation completed')
num_rows = results.shape[0]

我在model1_utilities.py的return语句前也添加了日志打印逻辑：logger.info("Completed function call")。
目前已在model1_utilities.py中完成logger包的导入，且在main.py中添加了所有依赖包的搜索路径，但未在model1_utilities.py中配置相同的路径，不确定该操作是否必要，无法定位报错根本原因。

报错信息

程序运行过程中抛出如下错误：

INFO:main:Completed function call                               
INFO:main:Calculation completed
22/06/16 15:04:36 ERROR Executor: Exception in task 0.0 in stage 451.0 (TID 908)
org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "/Users/user1/Desktop/code/models/model1/model1_utilities.py", line 3, in <module>
    import logger
ModuleNotFoundError: No module named 'logger'

        at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:555)
        at org.apache.spark.sql.execution.python.PythonArrowOutput$$anon$1.read(PythonArrowOutput.scala:101)
        at org.apache.spark.sql.execution.python.PythonArrowOutput$$anon$1.read(PythonArrowOutput.scala:50)
        at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:508)
        at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
        at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:491)
        at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)
        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage87.agg_doAggregateWithoutKey_0$(Unknown Source)
        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage87.processNext(Unknown Source)
        at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
        at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:759)
        at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)
        at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:140)
        at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)
        at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)
        at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)
        at org.apache.spark.scheduler.Task.run(Task.scala:131)
        at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:506)
        at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1462)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:509)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:750)
22/06/16 15:04:36 WARN TaskSetManager: Lost task 0.0 in stage 451.0 (TID 908) (10.0.0.162 executor driver): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "/Users/user1/Desktop/code/models/model1/model1_utilities.py", line 3, in <module>
    import logger
ModuleNotFoundError: No module named 'logger'

        at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:555)
        at org.apache.spark.sql.execution.python.PythonArrowOutput$$anon$1.read(PythonArrowOutput.scala:101)
        at org.apache.spark.sql.execution.python.PythonArrowOutput$$anon$1.read(PythonArrowOutput.scala:50)
        at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:508)
        at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
        at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:491)
        at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)
        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage87.agg_doAggregateWithoutKey_0$(Unknown Source)
        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage87.processNext(Unknown Source)
        at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
        at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:759)
        at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460)
        at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:140)
        at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)
        at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)
        at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)
        at org.apache.spark.scheduler.Task.run(Task.scala:131)
        at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:506)
        at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1462)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:509)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:750)

22/06/16 15:04:36 ERROR TaskSetManager: Task 0 in stage 451.0 failed 1 times; aborting job
ERROR:main:Program did not finish successfully
Traceback (most recent call last):
  File "/Users/user1/Desktop/code/models/model1/main.py", line 174, in <module>
    num_alerts = results.shape[0]
  File "/Users/user1/opt/anaconda3/envs/work/lib/python3.9/site-packages/pyspark/pandas/frame.py", line 7445, in shape
    return len(self), len(self.columns)
  File "/Users/user1/opt/anaconda3/envs/work/lib/python3.9/site-packages/pyspark/pandas/frame.py", line 11909, in __len__
    return self._internal.resolved_copy.spark_frame.count()
  File "/Users/user1/opt/anaconda3/envs/work/lib/python3.9/site-packages/pyspark/sql/dataframe.py", line 680, in count
    return int(self._jdf.count())
  File "/Users/user1/Desktop/work/tools/spark-3.2.1-bin-hadoop3.2/python/lib/py4j-0.10.9.3-src.zip/py4j/java_gateway.py", line 1321, in __call__
    return_value = get_return_value(
  File "/Users/user1/opt/anaconda3/envs/work/lib/python3.9/site-packages/pyspark/sql/utils.py", line 117, in deco
    raise converted from None
pyspark.sql.utils.PythonException: 
  An exception was thrown from the Python worker. Please see the stack trace below.
Traceback (most recent call last):
  File "/Users/user1/Desktop/code/models/model1/model1_utilities.py", line 3, in <module>
    import logger
ModuleNotFoundError: No module named 'logger'
问题根因

main.py中手动添加的模块搜索路径仅对Driver进程生效。Spark执行计算任务时，运行在Executor上的Python Worker进程使用独立的Python运行环境，不会继承Driver端临时修改的sys.path配置。
关键现象说明：日志中前两条打印成功，是因为对应逻辑在Driver端执行；当代码执行到results.shape[0]时会触发count Action算子，实际计算逻辑被分发到Executor上运行，此时Executor加载model1_utilities.py时无法找到项目根目录下的logger包，直接抛出导入错误。

解决方案

不需要在model1_utilities.py中重复配置路径，使用Spark原生的依赖分发机制即可解决问题，可选方案如下：

  • 方案1：初始化SparkSession时通过配置分发依赖
    初始化时将自定义依赖包路径加入spark.submit.pyFiles配置，Spark会自动将指定的文件/包分发到所有Executor的工作目录，保证导入逻辑正常：
    from pyspark.sql import SparkSession
import sys
import os

# 替换为本地logger包所在的项目根目录绝对路径
project_root = "/Users/user1/Desktop/code"
# 给Driver端添加模块搜索路径
sys.path.append(project_root)

spark = SparkSession.builder \
    .appName("model1_calculation") \
    # 多个依赖用逗号分隔，支持单个py文件、目录、zip/egg包
    .config("spark.submit.pyFiles", os.path.join(project_root, "logger")) \
    .getOrCreate()
    如果是通过spark-submit命令提交任务，直接添加--py-files参数传入依赖路径即可，无需在代码中写配置。
  • 方案2：打包依赖后统一分发
    当自定义依赖较多时，进入项目根目录将所有自定义包打包为压缩包：
    zip -r project_deps.zip logger/ another_package/
    后续将生成的project_deps.zip加入pyFiles配置或提交参数即可，维护成本更低。
  • 避坑提示：不要在分发到Executor执行的工具代码中硬编码绝对路径修改sys.path，集群不同节点的目录结构不一致会直接触发新的路径错误，使用Spark自带的依赖分发机制稳定性最高。

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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分)， 限时 9.9 元起

产品介绍相关文档
相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分)， 限时 9.9 元起

产品介绍相关文档
火山引擎 最新活动
全面开放
Doubao-Seedance-2.5全新模型升级

模型再升级,30秒超长叙事, 模态参考扩容

HOT
方舟Codingplan

模型自由，工具不限，最新支持 Deepseek-V4 系列、GLM-5.3 系列

HOT
方舟Agent plan

超全模态模型 × Harness 升级，最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列

HOT
新用户特惠专场

大模型19元起，Al应用9.9元畅享，新人首购爆款尽享优惠

最近更新时间：2026.08.30 06:45:42