Luigi: pyspark_runner.py can't un-pickle a PySparkTask sub-class

[neilisaac]

Using luigi 2.0.1 with python 2.7 and spark 1.5.2 on mapr 5.0.0, I'm unable to use PySparkTask directly.

Possibly related to #1185

Symptom

Traceback (most recent call last):
  File "/usr/local/lib/python2.7/dist-packages/luigi/worker.py", line 147, in run
    new_deps = self._run_get_new_deps()
  File "/usr/local/lib/python2.7/dist-packages/luigi/worker.py", line 98, in _run_get_new_deps
    task_gen = self.task.run()
  File "/usr/local/lib/python2.7/dist-packages/luigi/contrib/spark.py", line 340, in run
    super(PySparkTask, self).run()
  File "/usr/local/lib/python2.7/dist-packages/luigi/contrib/spark.py", line 250, in run
    raise SparkJobError('Spark job failed {0}'.format(repr(args)), out=stdout, err=stderr)

Traceback (most recent call last):
  File "/usr/local/lib/python2.7/dist-packages/luigi/contrib/pyspark_runner.py", line 57, in <module>
    PySparkRunner(*sys.argv[1:]).run()
  File "/usr/local/lib/python2.7/dist-packages/luigi/contrib/pyspark_runner.py", line 43, in __init__
    self.job = pickle.load(fd)
ImportError: No module named luigi_spark_test

Steps to reproduce

• Create luigi_spark_test.py containing:

import luigi
import luigi.contrib.spark

class DoNothingTask(luigi.contrib.spark.PySparkTask):
    def main(self, sc):
        pass

if __name__ == '__main__':
    luigi.run(main_task_cls=DoNothingTask)

• Run python2 luigi_spark_test.py
  

  Speculation

pyspark_runner.py is failing to un-pickle because it can't load the luigi_spark_test module.

PySparkTask uses pickle to serialize the state of the task, but pickle doesn't actually serialize the class's code. It only serializes the __dict__ (the state) of the object, and saves the name of the class used to recreate it. See https://docs.python.org/2/library/pickle.html#what-can-be-pickled-and-unpickled ("Note that none of the class’s code or data is pickled")

pyspark_runner.py can't import luigi_spark_test when it runs via spark-submit

Proof-of-concept workaround

Inheriting from this class (which overrides PySparkTask.run), I can get my job to work. Commented where it differs.

The basic idea is to copy luigi_spark_test.py into the same folder as pyspark_runner.py so that pyspark_runner can import luigi_spark_test during pickle.load()

import inspect
import os
import tempfile
import shutil

import luigi.contrib.spark
import luigi.contrib.pyspark_runner

class PySparkTaskHelper(luigi.contrib.spark.PySparkTask):

    def run(self):
        runner_file = luigi.contrib.pyspark_runner.__file__
        python_file = inspect.getfile(self.__class__)

        self.run_path = tempfile.mkdtemp(prefix=self.name)

        # copy pyspark_runner.py into the temp folder and use that copy instead
        self.app = os.path.join(self.run_path, 'pyspark_runner.py')
        shutil.copy(runner_file, os.path.join(self.run_path, 'pyspark_runner.py'))

        # copy the module containing the current class's definition
        # this likely won't work in the general case
        shutil.copy(python_file, os.path.join(self.run_path, python_file))

        # carry on as usual
        self.run_pickle = os.path.join(self.run_path, '.'.join([self.name.replace(' ', '_'), 'pickle']))
        with open(self.run_pickle, 'wb') as fd:
            self._dump(fd)
        try:
            # call SparkSubmitTask.run() instead of PySparkTask.run()
            luigi.contrib.spark.SparkSubmitTask.run(self)
        finally:
            shutil.rmtree(self.run_path)

This is just a rough hack to demonstrate the issue, not a proposed solution to be merged.

[jurriaan]

Having the same issue here

[erikbern]

that sounds really weird. can you troubleshoot this and submit a pull request? sorry I have never used the Spark functionality :)

[Tarrasch]

I haven't used PySpark either (I've only used subprocess calls to scala-spark). Thanks for sharing this issue so others can find it. Though you'll likely need to fix it yourself. :)

[ShT3ch]

My situation was worse: I had ImportError: No module named luigi because there is no luigi in global python context, it only in my vertualenv.

So, my solution is:

$ source internal_code/bin/activate
(internal_code) $ PYSPARK_DRIVER_PYTHON=$VIRTUAL_ENV/python PYTHONPATH=`pwd` luigi --module spark_trial STest

All happens because unpickling is in pyspark_runner.py. So, when you are submitting job by /usr/share/spark/spark-submit /home/.../internal_code/local/lib/python2.7/site-packages/luigi/contrib/pyspark_runner.py its executed by sys.env.getOrElse("PYSPARK_DRIVER_PYTHON", sys.env.getOrElse("PYSPARK_PYTHON", "python")) wrong python with wrong packages.

[mfcabrera]

Hi all, I know this is an old issue, but I am having the same problem as @ShT3ch. @Tarrasch, For this in particular wouldn't be enough to add an option to the runner (via [spark] section) to set PYSPARK_DRIVER_PYTHON environment variable to the right location, and the return it to the normal state. so the right Python executable is picked. I could create a PR for that.

[villanuevab]

I am experiencing the same issue; @neilisaac 's solution does not work in my case. I get the following:

1. Seems like the script did what it was supposed to do and placed both pyspark_runner.py and my_spark_task.py in the same /tmp/ folder:
   INFO: Running command: spark-submit --deploy-mode client --name MySparkTask --conf spark.yarn.driver.memoryOverhead=5000M --conf spark.yarn.executor.memoryOverhead=20G --driver-memory 15g --executor-memory 55g /tmp/MySparkTask8zRqZ0/pyspark_runner.py /tmp/MySparkTask8zRqZ0/MySparkTask.pickle
2. However, the module still cannot be found.
   ImportError: No module named my_spark_task

[ntim]

Hi there, the input error is thrown because the pickler does not find the module to load the class from. In the pull request I made, this issue is fixed for python 2.7 and python 3.6. Please try for yourself.

[stale[bot]]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. If closed, you may revisit when your time allows and reopen! Thank you for your contributions.