PySpark使用RDD reduce函数抛出Py4JJavaError问题求助

核心问题 #

报错关键信息：Cannot run program "python3": CreateProcess error=2, The system cannot find the file specified。Windows系统中Python默认可执行文件为python.exe，但Spark默认尝试调用python3，导致找不到程序引发报错。

环境版本 #

• Java：1.8.0_202
• Python：3.12.1
• PySpark：3.5.0

问题代码 #

import pandas as pd
from collections import Counter
from util import *
from pyspark import SparkConf, SparkContext

cols = ['sentiment', 'id', 'date', 'query_string', 'user', 'text']
df = pd.read_csv('file.csv', encoding='latin1', names=cols)

data = df['text'].tolist()

conf = SparkConf().setAppName('MapReduce').setMaster('local[4]')
sc = SparkContext(conf=conf)

dist = sc.parallelize(data)

def mapper(text):
    words = text.split()
    cleaned = map(clean_word, words)
    filtered = filter(word_not_in_stopwords, cleaned)
    return Counter(filtered)
    
def reducer(cnt1, cnt2):
    cnt1.update(cnt2)
    return cnt1

dist_mapped = dist.map(mapper)
dist_reduced = dist_mapped.reduce(reducer)

dist_data_reduced.most_common(5)

报错回溯 #

---------------------------------------------------------------------------
Py4JJavaError                             Traceback (most recent call last)
Cell In[10], line 2
      1 dist_mapped = dist.map(mapper)
----> 2 dist_reduced = dist_mapped.reduce(reducer)

File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\pyspark\rdd.py:1924, in RDD.reduce(self, f)
   1921         return
   1922     yield reduce(f, iterator, initial)
-> 1924 vals = self.mapPartitions(func).collect()
   1925 if vals:
   1926     return reduce(f, vals)

File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\pyspark\rdd.py:1833, in RDD.collect(self)
   1831 with SCCallSiteSync(self.context):
   1832     assert self.ctx._jvm is not None
-> 1833     sock_info = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd())
   1834 return list(_load_from_socket(sock_info, self._jrdd_deserializer))

File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\py4j\java_gateway.py:1322, in JavaMember.__call__(self, *args)
   1316 command = proto.CALL_COMMAND_NAME +\
   1317     self.command_header +\
   1318     args_command +\
   1319     proto.END_COMMAND_PART
   1321 answer = self.gateway_client.send_command(command)
-> 1322 return_value = get_return_value(
   1323     answer, self.gateway_client, self.target_id, self.name)
   1325 for temp_arg in temp_args:
   1326     if hasattr(temp_arg, "_detach"):

File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\py4j\protocol.py:326, in get_return_value(answer, gateway_client, target_id, name)
    324 value = OUTPUT_CONVERTER[type](answer[2:], gateway_client)
    325 if answer[1] == REFERENCE_TYPE:
-> 326     raise Py4JJavaError(
    327         "An error occurred while calling {0}{1}{2}.\n".
    328         format(target_id, ".", name), value)
    329 else:
    330     raise Py4JError(
    331         "An error occurred while calling {0}{1}{2}. Trace:\n{3}\n".
    332         format(target_id, ".", name, value))

Py4JJavaError: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.collectAndServe.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 2 in stage 0.0 failed 1 times, most recent failure: Lost task 2.0 in stage 0.0 (TID 2) (Corsair executor driver): java.io.IOException: Cannot run program "python3": CreateProcess error=2, The system cannot find the file specified
    at java.lang.ProcessBuilder.start(ProcessBuilder.java:1048)
    ...

系统环境变量 #

• HADOOP_HOME = C:\hadoop
• SPARK_HOME = %USERPROFILE%\AppData\Local\Programs\Python\Python312\Lib\site-packages\pyspark
• Path = %HADOOP_HOME%\bin
• Path = %SPARK_HOME%\bin

用户环境变量 #

• JDK_HOME = %JAVA_HOME%
• JRE_HOME = %JAVA_HOME%\jre
• Path = %JAVA_HOME%\lib
• Path = %JAVA_HOME%\jre\lib

解决方法 #

方法1：在Spark配置中指定Python路径 #

修改SparkConf，明确指定Python可执行文件：

conf = SparkConf().setAppName('MapReduce').setMaster('local[4]')
# 添加两行配置指定Python路径
conf.set("spark.executorEnv.PYSPARK_PYTHON", "python.exe")
conf.set("spark.driverEnv.PYSPARK_PYTHON", "python.exe")
sc = SparkContext(conf=conf)

方法2：添加系统环境变量 #

1. 打开「系统属性」→「高级」→「环境变量」
2. 新建系统变量：变量名PYSPARK_PYTHON，变量值python.exe
3. 重启终端或IDE，使环境变量生效

方法3：创建python3软链接（Windows） #

找到Python安装目录，复制python.exe并重命名为python3.exe，确保Python安装目录已加入系统Path。

额外检查 #

• 确认Python安装目录已添加到系统Path
• 终端输入python --version，验证可正常调用Python

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