Spark IndexError元组索引越界求助：Jupyter正常PyCharm(Flask)报错

问题详情 #

• 环境差异：相同代码在Jupyter Notebook可正常运行，但PyCharm的Flask API环境触发IndexError: tuple index out of range
• 已尝试无效方案：更换Python版本、更新cloudpickle、将Pandas DataFrame转换为Spark DataFrame
• 版本信息：Python 3.10.9，Spark 3.3.2

相关代码 #

params = {
    "spark.driver.memory": "16G",
    "spark.kryoserializer.buffer.max": "2000M",
    "spark.driver.maxResultSize": "2000M",
    "spark.ui.reverseProxyUrl": "http://localhost:4041"
}
spark = sparknlp_jsl.start(license_keys['SECRET'], params=params)
print("Spark NLP Version :")
spark_ins = spark
print("Spark NLP Version :", sparknlp.version())
print("Spark NLP_JSL Version :", sparknlp_jsl.version())
# spark_pipeline(spark,pipeline_all=clinical_profiling_pipeline_instance)


sample_text = [
    "The patient is a 60-years-old native american woman, diagnosed with liver cancer at age 58. She is a nonsmoker and has no other medical history.",
    "The patient is a 69 y.o. man with prostatic cancer. He was diagnosed at age 65, and he quit smoking at that time.",
    "She has no family history of cancer, and none in her family is a smoker."]

from pyspark.sql.types import StringType, IntegerType
from pyspark.sql.types import StringType, IntegerType
df = spark.createDataFrame(sample_text, StringType()).toDF('text')

报错堆栈 #

Traceback (most recent call last):
  File "C:\Users\Administrator\AppData\Local\Programs\Python\Python311\Lib\site-packages\pyspark\serializers.py", line 437, in dumps
    return cloudpickle.dumps(obj, pickle_protocol)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\Administrator\AppData\Local\Programs\Python\Python311\Lib\site-packages\pyspark\cloudpickle\cloudpickle_fast.py", line 72, in dumps
    cp.dump(obj)
  File "C:\Users\Administrator\AppData\Local\Programs\Python\Python311\Lib\site-packages\pyspark\cloudpickle\cloudpickle_fast.py", line 540, in dump
    return Pickler.dump(self, obj)
           ^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\Administrator\AppData\Local\Programs\Python\Python311\Lib\site-packages\pyspark\cloudpickle\cloudpickle_fast.py", line 630, in reducer_override
    return self._function_reduce(obj)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\Administrator\AppData\Local\Programs\Python\Python311\Lib\site-packages\pyspark\cloudpickle\cloudpickle_fast.py", line 503, in _function_reduce
    return self._dynamic_function_reduce(obj)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\Administrator\AppData\Local\Programs\Python\Python311\Lib\site-packages\pyspark\cloudpickle\cloudpickle_fast.py", line 484, in _dynamic_function_reduce
    state = _function_getstate(func)
            ^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\Administrator\AppData\Local\Programs\Python\Python311\Lib\site-packages\pyspark\cloudpickle\cloudpickle_fast.py", line 156, in _function_getstate
    f_globals_ref = _extract_code_globals(func.__code__)
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\Administrator\AppData\Local\Programs\Python\Python311\Lib\site-packages\pyspark\cloudpickle\cloudpickle.py", line 236, in _extract_code_globals
    out_names = {names[oparg] for _, oparg in _walk_global_ops(co)}
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\Administrator\AppData\Local\Programs\Python\Python311\Lib\site-packages\pyspark\cloudpickle\cloudpickle.py", line 236, in <setcomp>
    out_names = {names[oparg] for _, oparg in _walk_global_ops(co)}
                 ~~~~~^^^^^^^
IndexError: tuple index out of range

可能的解决方案 #

1. 修正Python解释器版本 #

报错堆栈显示使用的是Python 3.11的依赖包路径，但你声明的是Python 3.10.9：

• 打开PyCharm的「Project Settings → Python Interpreter」，确认选择的是Python 3.10.9的虚拟环境
• 执行「File → Invalidate Caches...」清理缓存后重启PyCharm

2. 封装Spark逻辑到独立模块 #

Flask的请求上下文会干扰函数全局变量引用，导致cloudpickle序列化失败：

• 将Spark初始化、DataFrame创建逻辑封装到独立的Python模块函数中，避免直接在Flask路由函数内执行
• 确保函数内部不引用Flask上下文相关变量

3. 指定兼容的cloudpickle版本 #

Spark 3.3.2自带的cloudpickle版本与Python 3.10存在兼容性问题，尝试安装指定版本：

pip install cloudpickle==2.2.1

4. 全局初始化Spark实例 #

不要在Flask路由内重复创建SparkContext，应在App启动时全局初始化一次：

# app.py 全局范围初始化Spark
params = {
    "spark.driver.memory": "16G",
    "spark.kryoserializer.buffer.max": "2000M",
    "spark.driver.maxResultSize": "2000M",
    "spark.ui.reverseProxyUrl": "http://localhost:4041"
}
spark = sparknlp_jsl.start(license_keys['SECRET'], params=params)

@app.route('/process-text')
def process_text():
    sample_text = [...]
    df = spark.createDataFrame(sample_text, StringType()).toDF('text')
    # 后续处理逻辑

5. 修改Spark序列化配置 #

强制使用Java序列化替代cloudpickle，避免序列化冲突：

params = {
    # 原有参数
    "spark.serializer": "org.apache.spark.serializer.JavaSerializer",
    "spark.driver.extraJavaOptions": "-Djava.io.tmpdir=/tmp"
}

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