title: Jupyter Notebook PySpark OSError [WinError 123] The filename, directory name, or volume label syntax is incorrect:
body: <p><b>System Configuration: </b>
  Operating System: Windows 10
  Python Version: 3.7
  Spark Version: 2.4.4
  SPARK_HOME: C:\spark\spark-2.4.4-bin-hadoop2.7</p>
  
  <p><b>Problem </b>
  I am using PySpark to do parallel computations on all the columns of a row in a dataframe. I convert my Pandas Dataframe to Spark dataframe. On the spark dataframe, the map transformation and collect action is performed. While, performing the collect operation the Py4J error with OSError pops up. The error arises in the import sklearn statement and in the trained classifier(ML Model).</p>
  
  <p><b> Code Snippet</b></p>
  
  <pre><code>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()
  </code></pre>
  
  <p><b> Error </b></p>
  
  <pre><code>---------------------------------------------------------------------------
  Py4JJavaError                             Traceback (most recent call last)
  &lt;ipython-input-33-0bfb9d088e2d&gt; in &lt;module&gt;
  ----&gt; 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 &lt;module&gt;
      from .forest import RandomForestClassifier
    File "C:\Users\rkagr\Anaconda3\lib\site-packages\sklearn\ensemble\forest.py", line 53, in &lt;module&gt;
      from ..metrics import r2_score
    File "C:\Users\rkagr\Anaconda3\lib\site-packages\sklearn\metrics\__init__.py", line 7, in &lt;module&gt;
      from .ranking import auc
    File "C:\Users\rkagr\Anaconda3\lib\site-packages\sklearn\metrics\ranking.py", line 35, in &lt;module&gt;
      from ..preprocessing import label_binarize
    File "C:\Users\rkagr\Anaconda3\lib\site-packages\sklearn\preprocessing\__init__.py", line 6, in &lt;module&gt;
      from ._function_transformer import FunctionTransformer
    File "C:\Users\rkagr\Anaconda3\lib\site-packages\sklearn\preprocessing\_function_transformer.py", line 5, in &lt;module&gt;
      from ..utils.testing import assert_allclose_dense_sparse
    File "C:\Users\rkagr\Anaconda3\lib\site-packages\sklearn\utils\testing.py", line 718, in &lt;module&gt;
      import pytest
    File "C:\Users\rkagr\Anaconda3\lib\site-packages\pytest.py", line 6, in &lt;module&gt;
      from _pytest.assertion import register_assert_rewrite
    File "C:\Users\rkagr\Anaconda3\lib\site-packages\_pytest\assertion\__init__.py", line 6, in &lt;module&gt;
      from _pytest.assertion import rewrite
    File "C:\Users\rkagr\Anaconda3\lib\site-packages\_pytest\assertion\rewrite.py", line 20, in &lt;module&gt;
      from _pytest.assertion import util
    File "C:\Users\rkagr\Anaconda3\lib\site-packages\_pytest\assertion\util.py", line 5, in &lt;module&gt;
      import _pytest._code
    File "C:\Users\rkagr\Anaconda3\lib\site-packages\_pytest\_code\__init__.py", line 2, in &lt;module&gt;
      from .code import Code  # noqa
    File "C:\Users\rkagr\Anaconda3\lib\site-packages\_pytest\_code\code.py", line 11, in &lt;module&gt;
      import pluggy
    File "C:\Users\rkagr\Anaconda3\lib\site-packages\pluggy\__init__.py", line 16, in &lt;module&gt;
      from .manager import PluginManager, PluginValidationError
    File "C:\Users\rkagr\Anaconda3\lib\site-packages\pluggy\manager.py", line 6, in &lt;module&gt;
      import importlib_metadata
    File "C:\Users\rkagr\Anaconda3\lib\site-packages\importlib_metadata\__init__.py", line 466, in &lt;module&gt;
      __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 &lt;genexpr&gt;
      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'
  </code></pre>
  
  <p><b> MCVE </b>
  This MCVE defines the function to just return the same input row as dictionary whereas the original code returns a dictionary after some processing.</p>
  
  <pre><code>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)
  </code></pre>