Azure Databricks中PySpark报找不到pandas._libs.arrays模块错误求助

问题场景 #

在Azure Databricks Notebook运行以下PySpark代码时，执行pd.concat(rdd_s.collect())触发依赖错误：

rdd_s = sc.parallelize(input_dict)
rdd_s = rdd_s.map(lambda day: function(day))
results = pd.concat(rdd_s.collect()) # 触发错误的行

错误信息：

org.apache.spark.SparkException: Job aborted due to stage failure: Task 2 in stage 19742.0 failed 4 times, most recent failure: Lost task 2.3 in stage 19742.0 (TID 80976) (10.5.0.15 executor 42): org.apache.spark.api.python.PythonException: 'ModuleNotFoundError: No module named 'pandas._libs.arrays''

核心问题是Driver与Executor节点的Pandas版本不一致，或Executor节点未正确安装对应版本的Pandas。

解决步骤 #

1. 统一集群节点的Pandas版本 #

• 首先在Notebook中执行以下代码，确认Driver节点的Pandas版本：

  import pandas as pd
  print(f"Driver Pandas版本: {pd.__version__}")
  

• 确保所有Executor节点安装相同版本的Pandas：
  • 方式一：通过Databricks集群的「库」管理界面，添加指定版本的Pandas库（选择PyPI源，输入pandas==x.x.x，x.x.x为Driver查到的版本）
  • 方式二：在Notebook开头执行以下命令，强制安装指定版本：

    %pip install pandas==x.x.x --force-reinstall
    

• 安装完成后重启集群，确保依赖同步到所有节点。

2. 验证Executor节点的Pandas环境 #

执行以下代码，检查所有Executor节点的Pandas版本是否与Driver一致：

def check_executor_pandas():
    import pandas as pd
    return pd.__version__

# 提交分布式任务验证
executor_versions = sc.parallelize(range(4)).map(lambda x: check_executor_pandas()).distinct().collect()
print(f"Executor节点Pandas版本: {executor_versions}")

如果输出的版本列表只有一个且与Driver版本一致，说明环境同步成功。

3. 优化代码减少跨节点依赖传递 #

如果依赖同步仍有问题，可以修改代码逻辑，避免在Executor端返回Pandas对象：

# 修改map函数，返回原生数据结构（如列表/字典）
def process_day(day):
    df = function(day)
    return df.to_dict('records') # 将DataFrame转为字典列表

rdd_s = sc.parallelize(input_dict)
rdd_s = rdd_s.map(lambda day: process_day(day))

# 在Driver端合并数据为DataFrame
results = pd.DataFrame([item for sublist in rdd_s.collect() for item in sublist])

4. 检查集群Python环境一致性 #

• 确认集群使用的Python环境（Databricks Runtime自带或自定义conda环境），保证Driver与Executor使用同一套Python解释器
• 若使用自定义conda环境，需确保环境已通过Databricks集群配置同步到所有Executor节点

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