已解决：Jupyter Notebook PySpark OSError [WinError 123] 文件名、目录名或卷标语法不正确：答案

【问题标题】：Jupyter Notebook PySpark OSError [WinError 123] The filename, directory name, or volume label syntax is incorrect:已解决：Jupyter Notebook PySpark OSError [WinError 123] 文件名、目录名或卷标语法不正确：
【发布时间】：2020-03-24 02:13:23
【问题描述】：

系统配置： 操作系统：Windows 10 Python版本：3.7 火花版本：2.4.4 SPARK_HOME: C:\spark\spark-2.4.4-bin-hadoop2.7

问题我正在使用 PySpark 对数据帧中一行的所有列进行并行计算。我将我的 Pandas 数据框转换为 Spark 数据框。在 spark 数据帧上，执行映射转换和收集操作。同时，执行收集操作时会弹出带有 OSError 的 Py4J 错误。错误出现在 import sklearn 语句和经过训练的分类器（ML 模型）中。

代码片段

from sklearn.neural_network.multilayer_perceptron import MLPClassifier
classifier=MLPClassifier()
classifier.fit(x_train, y_train)

def func1(rows,trained_model=classifier):
    items = rows.asDict()
    row = pd.Series(items)
    output = func2(row,trained_model) # Consumes pandas series in other file having import sklearn statement
    return output

spdf=spark.createDataFrame(pandasDF)
result=spdf.rdd.map(lambda row:func1(row)).collect()

错误

---------------------------------------------------------------------------
Py4JJavaError                             Traceback (most recent call last)
<ipython-input-33-0bfb9d088e2d> in <module>
----> 1 result=spdf.rdd.map(lambda row:clusterCreation(row)).collect()
      2 print(type(result))
.
.
.
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 2.0 failed 1 times, most recent failure: Lost task 2.0 in stage 2.0 (TID 5, localhost, executor driver): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "C:\spark\spark-2.4.4-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\worker.py", line 364, in main
  File "C:\spark\spark-2.4.4-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\worker.py", line 71, in read_command
  File "C:\spark\spark-2.4.4-bin-hadoop2.7\python\lib\pyspark.zip\pyspark\serializers.py", line 580, in loads
    return pickle.loads(obj, encoding=encoding)
.
.
.
 File "C:\Users\rkagr\Anaconda3\lib\site-packages\sklearn\ensemble\__init__.py", line 7, in <module>
    from .forest import RandomForestClassifier
  File "C:\Users\rkagr\Anaconda3\lib\site-packages\sklearn\ensemble\forest.py", line 53, in <module>
    from ..metrics import r2_score
  File "C:\Users\rkagr\Anaconda3\lib\site-packages\sklearn\metrics\__init__.py", line 7, in <module>
    from .ranking import auc
  File "C:\Users\rkagr\Anaconda3\lib\site-packages\sklearn\metrics\ranking.py", line 35, in <module>
    from ..preprocessing import label_binarize
  File "C:\Users\rkagr\Anaconda3\lib\site-packages\sklearn\preprocessing\__init__.py", line 6, in <module>
    from ._function_transformer import FunctionTransformer
  File "C:\Users\rkagr\Anaconda3\lib\site-packages\sklearn\preprocessing\_function_transformer.py", line 5, in <module>
    from ..utils.testing import assert_allclose_dense_sparse
  File "C:\Users\rkagr\Anaconda3\lib\site-packages\sklearn\utils\testing.py", line 718, in <module>
    import pytest
  File "C:\Users\rkagr\Anaconda3\lib\site-packages\pytest.py", line 6, in <module>
    from _pytest.assertion import register_assert_rewrite
  File "C:\Users\rkagr\Anaconda3\lib\site-packages\_pytest\assertion\__init__.py", line 6, in <module>
    from _pytest.assertion import rewrite
  File "C:\Users\rkagr\Anaconda3\lib\site-packages\_pytest\assertion\rewrite.py", line 20, in <module>
    from _pytest.assertion import util
  File "C:\Users\rkagr\Anaconda3\lib\site-packages\_pytest\assertion\util.py", line 5, in <module>
    import _pytest._code
  File "C:\Users\rkagr\Anaconda3\lib\site-packages\_pytest\_code\__init__.py", line 2, in <module>
    from .code import Code  # noqa
  File "C:\Users\rkagr\Anaconda3\lib\site-packages\_pytest\_code\code.py", line 11, in <module>
    import pluggy
  File "C:\Users\rkagr\Anaconda3\lib\site-packages\pluggy\__init__.py", line 16, in <module>
    from .manager import PluginManager, PluginValidationError
  File "C:\Users\rkagr\Anaconda3\lib\site-packages\pluggy\manager.py", line 6, in <module>
    import importlib_metadata
  File "C:\Users\rkagr\Anaconda3\lib\site-packages\importlib_metadata\__init__.py", line 466, in <module>
    __version__ = version(__name__)
  File "C:\Users\rkagr\Anaconda3\lib\site-packages\importlib_metadata\__init__.py", line 433, in version
    return distribution(package).version
  File "C:\Users\rkagr\Anaconda3\lib\site-packages\importlib_metadata\__init__.py", line 406, in distribution
    return Distribution.from_name(package)
  File "C:\Users\rkagr\Anaconda3\lib\site-packages\importlib_metadata\__init__.py", line 176, in from_name
    dist = next(dists, None)
  File "C:\Users\rkagr\Anaconda3\lib\site-packages\importlib_metadata\__init__.py", line 362, in <genexpr>
    for path in map(cls._switch_path, paths)
  File "C:\Users\rkagr\Anaconda3\lib\site-packages\importlib_metadata\__init__.py", line 377, in _search_path
    if not root.is_dir():
  File "C:\Users\rkagr\Anaconda3\lib\pathlib.py", line 1351, in is_dir
    return S_ISDIR(self.stat().st_mode)
  File "C:\Users\rkagr\Anaconda3\lib\pathlib.py", line 1161, in stat
    return self._accessor.stat(self)
OSError: [WinError 123] The filename, directory name, or volume label syntax is incorrect: 'C:\\C:\\spark\\spark-2.4.4-bin-hadoop2.7\\jars\\spark-core_2.11-2.4.4.jar'

MCVE 此 MCVE 将函数定义为只返回与字典相同的输入行，而原始代码在经过一些处理后返回字典。

import findspark

findspark.init()
findspark.find()

import pyspark
from pyspark import SparkContext, SparkConf
from pyspark.sql import SparkSession
conf = SparkConf().setAppName('MRC').setMaster('local[2]')
sc = SparkContext.getOrCreate(conf=conf)
spark = SparkSession.builder.getOrCreate()

import sklearn
import sklearn.datasets
import sklearn.model_selection
import sklearn.ensemble

iris = sklearn.datasets.load_iris()
train, test, labels_train, labels_test = sklearn.model_selection.train_test_split(iris.data, iris.target, train_size=0.80)

classifier = sklearn.ensemble.RandomForestClassifier()
classifier.fit(train, labels_train)

import pickle
path = './random_classifier.mdl'
pickle.dump(classifier, open(path,'wb'))

import pandas as pd
pddf=pd.DataFrame(test)
spdf=spark.createDataFrame(pddf)

def clusterCreation(rows,classifier_path):
    items = rows.asDict()
    row = pd.Series(items)
    with open(classifier_path,'rb') as fp:
        classifier = pickle.load(fp)
        print(classifier)
    return items

result=spdf.rdd.map(lambda row:clusterCreation(row,classifier_path=path)).collect()
print(result)

【问题讨论】：

嗨！卷标确实不正确。请注意，它指的是C:\\C:\\ 。其次，如果您可以使用 Pandas Dataframe 训练模型，为什么不继续使用 Pandas 进行映射（使用pd.DataFrame.apply）？这样，您就不会尝试腌制经过训练的模型，这可能不起作用（在这种情况下，使用 joblib 可能会更好地为您服务）。
是的，卷标貌似不对，但关键是我在整个代码中都没有提到卷标，它是隐含的。我想使用所有可用内核进行并行计算以加快执行速度，因此我使用 PySpark 而不是继续使用 Pandas pd.DataFrame.apply。
从堆栈跟踪中，我得到的印象是您的分类器不会腌制。与其通过映射函数传递它，不如让每个分区（从磁盘）加载经过训练的模型。如果这没有帮助，请提供mcve。
嘿@OliverW。！我尝试了你的建议，但我通过 sklearn 包得到了同样的错误。我正在用 mcve 更新发布的问题。请检查一下，让我知道我的错误。
您的 MCVE 在我的机器上运行（Ubuntu 18.04）。您使用的是哪个版本的 Spark 和 sklearn（在 pyspark 导入后运行 print(pyspark.__version__)）？你可以重新安装 Spark 吗？顺便说一句，对你的 MCVE 表示敬意：我很少看到有人在被要求提供一个如此好的 MCVE 之后。 ????

标签： python pyspark anaconda rdd sklearn-pandas

【解决方案1】：

我遇到了同样的文件路径包含C:\\C:\\的问题。我在https://github.com/Ibotta/sk-dist/issues/30 中发现了一个讨论，这表明这可能是pytest 在scikit-learn 中使用的问题。该问题在scikit-learn 0.21.3 版中报告。我将我的scikit-learn 包升级到 0.22.1（通过升级到 Anaconda 2020.02）并且错误消失了。

我的环境是 Windows 10、Spark 2.4.5、Anaconda 2020.02（其中包含 scikit-learn 0.22.1）。请注意，旧的 Anaconda 版本 2019.10 包含 scikit-learn 版本 0.21.3。

【讨论】：