We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
PySparkTask fails if name contains invalid characters
PySparkTask derives run_path and run_pickle from name property, but it do not escape special characters:
1. '/' in name causes FileNotFoundError raised by mkdtemp
2. ',' is separator of files in spark-submit '--files' arg, and run_pickle is passed to 'files' arg in cluster mode.
3. ':' causes java.net.URISyntaxException

To reproduce: create PySparkTask subclass with name property containing above special characters and try to run it.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `luigi/contrib/spark.py`
Content:
```
1 # -*- coding: utf-8 -*-
2 #
3 # Copyright 2012-2015 Spotify AB
4 #
5 # Licensed under the Apache License, Version 2.0 (the "License");
6 # you may not use this file except in compliance with the License.
7 # You may obtain a copy of the License at
8 #
9 # http://www.apache.org/licenses/LICENSE-2.0
10 #
11 # Unless required by applicable law or agreed to in writing, software
12 # distributed under the License is distributed on an "AS IS" BASIS,
13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
16 #
17 
18 import logging
19 import os
20 import sys
21 import tempfile
22 import shutil
23 import importlib
24 import tarfile
25 import inspect
26 try:
27     import cPickle as pickle
28 except ImportError:
29     import pickle
30 
31 from luigi import six
32 from luigi.contrib.external_program import ExternalProgramTask
33 from luigi import configuration
34 
35 logger = logging.getLogger('luigi-interface')
36 
37 
38 class SparkSubmitTask(ExternalProgramTask):
39     """
40     Template task for running a Spark job
41 
42     Supports running jobs on Spark local, standalone, Mesos or Yarn
43 
44     See http://spark.apache.org/docs/latest/submitting-applications.html
45     for more information
46 
47     """
48 
49     # Application (.jar or .py file)
50     name = None
51     entry_class = None
52     app = None
53 
54     # Only log stderr if spark fails (since stderr is normally quite verbose)
55     always_log_stderr = False
56     # Spark applications write its logs into stderr
57     stream_for_searching_tracking_url = 'stderr'
58 
59     def run(self):
60         if self.deploy_mode == "cluster":
61             # in cluster mode client only receives application status once a period of time
62             self.tracking_url_pattern = r"tracking URL: (https?://.*)\s"
63         else:
64             self.tracking_url_pattern = r"Bound (?:.*) to (?:.*), and started at (https?://.*)\s"
65 
66         super(SparkSubmitTask, self).run()
67 
68     def app_options(self):
69         """
70         Subclass this method to map your task parameters to the app's arguments
71 
72         """
73         return []
74 
75     @property
76     def spark_version(self):
77         return "spark"
78 
79     @property
80     def spark_submit(self):
81         return configuration.get_config().get(self.spark_version, 'spark-submit', 'spark-submit')
82 
83     @property
84     def master(self):
85         return configuration.get_config().get(self.spark_version, "master", None)
86 
87     @property
88     def deploy_mode(self):
89         return configuration.get_config().get(self.spark_version, "deploy-mode", None)
90 
91     @property
92     def jars(self):
93         return self._list_config(configuration.get_config().get(self.spark_version, "jars", None))
94 
95     @property
96     def packages(self):
97         return self._list_config(configuration.get_config().get(
98             self.spark_version, "packages", None))
99 
100     @property
101     def py_files(self):
102         return self._list_config(configuration.get_config().get(
103             self.spark_version, "py-files", None))
104 
105     @property
106     def files(self):
107         return self._list_config(configuration.get_config().get(self.spark_version, "files", None))
108 
109     @property
110     def conf(self):
111         return self._dict_config(configuration.get_config().get(self.spark_version, "conf", None))
112 
113     @property
114     def properties_file(self):
115         return configuration.get_config().get(self.spark_version, "properties-file", None)
116 
117     @property
118     def driver_memory(self):
119         return configuration.get_config().get(self.spark_version, "driver-memory", None)
120 
121     @property
122     def driver_java_options(self):
123         return configuration.get_config().get(self.spark_version, "driver-java-options", None)
124 
125     @property
126     def driver_library_path(self):
127         return configuration.get_config().get(self.spark_version, "driver-library-path", None)
128 
129     @property
130     def driver_class_path(self):
131         return configuration.get_config().get(self.spark_version, "driver-class-path", None)
132 
133     @property
134     def executor_memory(self):
135         return configuration.get_config().get(self.spark_version, "executor-memory", None)
136 
137     @property
138     def driver_cores(self):
139         return configuration.get_config().get(self.spark_version, "driver-cores", None)
140 
141     @property
142     def supervise(self):
143         return bool(configuration.get_config().get(self.spark_version, "supervise", False))
144 
145     @property
146     def total_executor_cores(self):
147         return configuration.get_config().get(self.spark_version, "total-executor-cores", None)
148 
149     @property
150     def executor_cores(self):
151         return configuration.get_config().get(self.spark_version, "executor-cores", None)
152 
153     @property
154     def queue(self):
155         return configuration.get_config().get(self.spark_version, "queue", None)
156 
157     @property
158     def num_executors(self):
159         return configuration.get_config().get(self.spark_version, "num-executors", None)
160 
161     @property
162     def archives(self):
163         return self._list_config(configuration.get_config().get(
164             self.spark_version, "archives", None))
165 
166     @property
167     def hadoop_conf_dir(self):
168         return configuration.get_config().get(self.spark_version, "hadoop-conf-dir", None)
169 
170     def get_environment(self):
171         env = os.environ.copy()
172         hadoop_conf_dir = self.hadoop_conf_dir
173         if hadoop_conf_dir:
174             env['HADOOP_CONF_DIR'] = hadoop_conf_dir
175         return env
176 
177     def program_environment(self):
178         return self.get_environment()
179 
180     def program_args(self):
181         return self.spark_command() + self.app_command()
182 
183     def spark_command(self):
184         command = [self.spark_submit]
185         command += self._text_arg('--master', self.master)
186         command += self._text_arg('--deploy-mode', self.deploy_mode)
187         command += self._text_arg('--name', self.name)
188         command += self._text_arg('--class', self.entry_class)
189         command += self._list_arg('--jars', self.jars)
190         command += self._list_arg('--packages', self.packages)
191         command += self._list_arg('--py-files', self.py_files)
192         command += self._list_arg('--files', self.files)
193         command += self._list_arg('--archives', self.archives)
194         command += self._dict_arg('--conf', self.conf)
195         command += self._text_arg('--properties-file', self.properties_file)
196         command += self._text_arg('--driver-memory', self.driver_memory)
197         command += self._text_arg('--driver-java-options', self.driver_java_options)
198         command += self._text_arg('--driver-library-path', self.driver_library_path)
199         command += self._text_arg('--driver-class-path', self.driver_class_path)
200         command += self._text_arg('--executor-memory', self.executor_memory)
201         command += self._text_arg('--driver-cores', self.driver_cores)
202         command += self._flag_arg('--supervise', self.supervise)
203         command += self._text_arg('--total-executor-cores', self.total_executor_cores)
204         command += self._text_arg('--executor-cores', self.executor_cores)
205         command += self._text_arg('--queue', self.queue)
206         command += self._text_arg('--num-executors', self.num_executors)
207         return command
208 
209     def app_command(self):
210         if not self.app:
211             raise NotImplementedError("subclass should define an app (.jar or .py file)")
212         return [self.app] + self.app_options()
213 
214     def _list_config(self, config):
215         if config and isinstance(config, six.string_types):
216             return list(map(lambda x: x.strip(), config.split(',')))
217 
218     def _dict_config(self, config):
219         if config and isinstance(config, six.string_types):
220             return dict(map(lambda i: i.split('=', 1), config.split('|')))
221 
222     def _text_arg(self, name, value):
223         if value:
224             return [name, value]
225         return []
226 
227     def _list_arg(self, name, value):
228         if value and isinstance(value, (list, tuple)):
229             return [name, ','.join(value)]
230         return []
231 
232     def _dict_arg(self, name, value):
233         command = []
234         if value and isinstance(value, dict):
235             for prop, value in value.items():
236                 command += [name, '{0}={1}'.format(prop, value)]
237         return command
238 
239     def _flag_arg(self, name, value):
240         if value:
241             return [name]
242         return []
243 
244 
245 class PySparkTask(SparkSubmitTask):
246     """
247     Template task for running an inline PySpark job
248 
249     Simply implement the ``main`` method in your subclass
250 
251     You can optionally define package names to be distributed to the cluster
252     with ``py_packages`` (uses luigi's global py-packages configuration by default)
253 
254     """
255 
256     # Path to the pyspark program passed to spark-submit
257     app = os.path.join(os.path.dirname(__file__), 'pyspark_runner.py')
258 
259     @property
260     def name(self):
261         return self.__class__.__name__
262 
263     @property
264     def py_packages(self):
265         packages = configuration.get_config().get('spark', 'py-packages', None)
266         if packages:
267             return map(lambda s: s.strip(), packages.split(','))
268 
269     @property
270     def files(self):
271         if self.deploy_mode == "cluster":
272             return [self.run_pickle]
273 
274     def setup(self, conf):
275         """
276         Called by the pyspark_runner with a SparkConf instance that will be used to instantiate the SparkContext
277 
278         :param conf: SparkConf
279         """
280 
281     def setup_remote(self, sc):
282         self._setup_packages(sc)
283 
284     def main(self, sc, *args):
285         """
286         Called by the pyspark_runner with a SparkContext and any arguments returned by ``app_options()``
287 
288         :param sc: SparkContext
289         :param args: arguments list
290         """
291         raise NotImplementedError("subclass should define a main method")
292 
293     def app_command(self):
294         if self.deploy_mode == "cluster":
295             pickle_loc = os.path.basename(self.run_pickle)
296         else:
297             pickle_loc = self.run_pickle
298         return [self.app, pickle_loc] + self.app_options()
299 
300     def run(self):
301         self.run_path = tempfile.mkdtemp(prefix=self.name)
302         self.run_pickle = os.path.join(self.run_path, '.'.join([self.name.replace(' ', '_'), 'pickle']))
303         with open(self.run_pickle, 'wb') as fd:
304             # Copy module file to run path.
305             module_path = os.path.abspath(inspect.getfile(self.__class__))
306             shutil.copy(module_path, os.path.join(self.run_path, '.'))
307             self._dump(fd)
308         try:
309             super(PySparkTask, self).run()
310         finally:
311             shutil.rmtree(self.run_path)
312 
313     def _dump(self, fd):
314         with self.no_unpicklable_properties():
315             if self.__module__ == '__main__':
316                 d = pickle.dumps(self)
317                 module_name = os.path.basename(sys.argv[0]).rsplit('.', 1)[0]
318                 d = d.replace(b'c__main__', b'c' + module_name.encode('ascii'))
319                 fd.write(d)
320             else:
321                 pickle.dump(self, fd)
322 
323     def _setup_packages(self, sc):
324         """
325         This method compresses and uploads packages to the cluster
326 
327         """
328         packages = self.py_packages
329         if not packages:
330             return
331         for package in packages:
332             mod = importlib.import_module(package)
333             try:
334                 mod_path = mod.__path__[0]
335             except AttributeError:
336                 mod_path = mod.__file__
337             tar_path = os.path.join(self.run_path, package + '.tar.gz')
338             tar = tarfile.open(tar_path, "w:gz")
339             tar.add(mod_path, os.path.basename(mod_path))
340             tar.close()
341             sc.addPyFile(tar_path)
342
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/luigi/contrib/spark.py b/luigi/contrib/spark.py
--- a/luigi/contrib/spark.py
+++ b/luigi/contrib/spark.py
@@ -17,6 +17,7 @@
 
 import logging
 import os
+import re
 import sys
 import tempfile
 import shutil
@@ -298,8 +299,9 @@
         return [self.app, pickle_loc] + self.app_options()
 
     def run(self):
-        self.run_path = tempfile.mkdtemp(prefix=self.name)
-        self.run_pickle = os.path.join(self.run_path, '.'.join([self.name.replace(' ', '_'), 'pickle']))
+        path_name_fragment = re.sub(r'[^\w]', '_', self.name)
+        self.run_path = tempfile.mkdtemp(prefix=path_name_fragment)
+        self.run_pickle = os.path.join(self.run_path, '.'.join([path_name_fragment, 'pickle']))
         with open(self.run_pickle, 'wb') as fd:
             # Copy module file to run path.
             module_path = os.path.abspath(inspect.getfile(self.__class__))

{"golden_diff": "diff --git a/luigi/contrib/spark.py b/luigi/contrib/spark.py\n--- a/luigi/contrib/spark.py\n+++ b/luigi/contrib/spark.py\n@@ -17,6 +17,7 @@\n \n import logging\n import os\n+import re\n import sys\n import tempfile\n import shutil\n@@ -298,8 +299,9 @@\n         return [self.app, pickle_loc] + self.app_options()\n \n     def run(self):\n-        self.run_path = tempfile.mkdtemp(prefix=self.name)\n-        self.run_pickle = os.path.join(self.run_path, '.'.join([self.name.replace(' ', '_'), 'pickle']))\n+        path_name_fragment = re.sub(r'[^\\w]', '_', self.name)\n+        self.run_path = tempfile.mkdtemp(prefix=path_name_fragment)\n+        self.run_pickle = os.path.join(self.run_path, '.'.join([path_name_fragment, 'pickle']))\n         with open(self.run_pickle, 'wb') as fd:\n             # Copy module file to run path.\n             module_path = os.path.abspath(inspect.getfile(self.__class__))\n", "issue": "PySparkTask fails if name contains invalid characters\nPySparkTask derives run_path and run_pickle from name property, but it do not escape special characters:\r\n1. '/' in name causes FileNotFoundError raised by mkdtemp\r\n2. ',' is separator of files in spark-submit '--files' arg, and run_pickle is passed to 'files' arg in cluster mode.\r\n3. ':' causes java.net.URISyntaxException\r\n\r\nTo reproduce: create PySparkTask subclass with name property containing above special characters and try to run it.\r\n\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\n#\n# Copyright 2012-2015 Spotify AB\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n#\n\nimport logging\nimport os\nimport sys\nimport tempfile\nimport shutil\nimport importlib\nimport tarfile\nimport inspect\ntry:\n    import cPickle as pickle\nexcept ImportError:\n    import pickle\n\nfrom luigi import six\nfrom luigi.contrib.external_program import ExternalProgramTask\nfrom luigi import configuration\n\nlogger = logging.getLogger('luigi-interface')\n\n\nclass SparkSubmitTask(ExternalProgramTask):\n    \"\"\"\n    Template task for running a Spark job\n\n    Supports running jobs on Spark local, standalone, Mesos or Yarn\n\n    See http://spark.apache.org/docs/latest/submitting-applications.html\n    for more information\n\n    \"\"\"\n\n    # Application (.jar or .py file)\n    name = None\n    entry_class = None\n    app = None\n\n    # Only log stderr if spark fails (since stderr is normally quite verbose)\n    always_log_stderr = False\n    # Spark applications write its logs into stderr\n    stream_for_searching_tracking_url = 'stderr'\n\n    def run(self):\n        if self.deploy_mode == \"cluster\":\n            # in cluster mode client only receives application status once a period of time\n            self.tracking_url_pattern = r\"tracking URL: (https?://.*)\\s\"\n        else:\n            self.tracking_url_pattern = r\"Bound (?:.*) to (?:.*), and started at (https?://.*)\\s\"\n\n        super(SparkSubmitTask, self).run()\n\n    def app_options(self):\n        \"\"\"\n        Subclass this method to map your task parameters to the app's arguments\n\n        \"\"\"\n        return []\n\n    @property\n    def spark_version(self):\n        return \"spark\"\n\n    @property\n    def spark_submit(self):\n        return configuration.get_config().get(self.spark_version, 'spark-submit', 'spark-submit')\n\n    @property\n    def master(self):\n        return configuration.get_config().get(self.spark_version, \"master\", None)\n\n    @property\n    def deploy_mode(self):\n        return configuration.get_config().get(self.spark_version, \"deploy-mode\", None)\n\n    @property\n    def jars(self):\n        return self._list_config(configuration.get_config().get(self.spark_version, \"jars\", None))\n\n    @property\n    def packages(self):\n        return self._list_config(configuration.get_config().get(\n            self.spark_version, \"packages\", None))\n\n    @property\n    def py_files(self):\n        return self._list_config(configuration.get_config().get(\n            self.spark_version, \"py-files\", None))\n\n    @property\n    def files(self):\n        return self._list_config(configuration.get_config().get(self.spark_version, \"files\", None))\n\n    @property\n    def conf(self):\n        return self._dict_config(configuration.get_config().get(self.spark_version, \"conf\", None))\n\n    @property\n    def properties_file(self):\n        return configuration.get_config().get(self.spark_version, \"properties-file\", None)\n\n    @property\n    def driver_memory(self):\n        return configuration.get_config().get(self.spark_version, \"driver-memory\", None)\n\n    @property\n    def driver_java_options(self):\n        return configuration.get_config().get(self.spark_version, \"driver-java-options\", None)\n\n    @property\n    def driver_library_path(self):\n        return configuration.get_config().get(self.spark_version, \"driver-library-path\", None)\n\n    @property\n    def driver_class_path(self):\n        return configuration.get_config().get(self.spark_version, \"driver-class-path\", None)\n\n    @property\n    def executor_memory(self):\n        return configuration.get_config().get(self.spark_version, \"executor-memory\", None)\n\n    @property\n    def driver_cores(self):\n        return configuration.get_config().get(self.spark_version, \"driver-cores\", None)\n\n    @property\n    def supervise(self):\n        return bool(configuration.get_config().get(self.spark_version, \"supervise\", False))\n\n    @property\n    def total_executor_cores(self):\n        return configuration.get_config().get(self.spark_version, \"total-executor-cores\", None)\n\n    @property\n    def executor_cores(self):\n        return configuration.get_config().get(self.spark_version, \"executor-cores\", None)\n\n    @property\n    def queue(self):\n        return configuration.get_config().get(self.spark_version, \"queue\", None)\n\n    @property\n    def num_executors(self):\n        return configuration.get_config().get(self.spark_version, \"num-executors\", None)\n\n    @property\n    def archives(self):\n        return self._list_config(configuration.get_config().get(\n            self.spark_version, \"archives\", None))\n\n    @property\n    def hadoop_conf_dir(self):\n        return configuration.get_config().get(self.spark_version, \"hadoop-conf-dir\", None)\n\n    def get_environment(self):\n        env = os.environ.copy()\n        hadoop_conf_dir = self.hadoop_conf_dir\n        if hadoop_conf_dir:\n            env['HADOOP_CONF_DIR'] = hadoop_conf_dir\n        return env\n\n    def program_environment(self):\n        return self.get_environment()\n\n    def program_args(self):\n        return self.spark_command() + self.app_command()\n\n    def spark_command(self):\n        command = [self.spark_submit]\n        command += self._text_arg('--master', self.master)\n        command += self._text_arg('--deploy-mode', self.deploy_mode)\n        command += self._text_arg('--name', self.name)\n        command += self._text_arg('--class', self.entry_class)\n        command += self._list_arg('--jars', self.jars)\n        command += self._list_arg('--packages', self.packages)\n        command += self._list_arg('--py-files', self.py_files)\n        command += self._list_arg('--files', self.files)\n        command += self._list_arg('--archives', self.archives)\n        command += self._dict_arg('--conf', self.conf)\n        command += self._text_arg('--properties-file', self.properties_file)\n        command += self._text_arg('--driver-memory', self.driver_memory)\n        command += self._text_arg('--driver-java-options', self.driver_java_options)\n        command += self._text_arg('--driver-library-path', self.driver_library_path)\n        command += self._text_arg('--driver-class-path', self.driver_class_path)\n        command += self._text_arg('--executor-memory', self.executor_memory)\n        command += self._text_arg('--driver-cores', self.driver_cores)\n        command += self._flag_arg('--supervise', self.supervise)\n        command += self._text_arg('--total-executor-cores', self.total_executor_cores)\n        command += self._text_arg('--executor-cores', self.executor_cores)\n        command += self._text_arg('--queue', self.queue)\n        command += self._text_arg('--num-executors', self.num_executors)\n        return command\n\n    def app_command(self):\n        if not self.app:\n            raise NotImplementedError(\"subclass should define an app (.jar or .py file)\")\n        return [self.app] + self.app_options()\n\n    def _list_config(self, config):\n        if config and isinstance(config, six.string_types):\n            return list(map(lambda x: x.strip(), config.split(',')))\n\n    def _dict_config(self, config):\n        if config and isinstance(config, six.string_types):\n            return dict(map(lambda i: i.split('=', 1), config.split('|')))\n\n    def _text_arg(self, name, value):\n        if value:\n            return [name, value]\n        return []\n\n    def _list_arg(self, name, value):\n        if value and isinstance(value, (list, tuple)):\n            return [name, ','.join(value)]\n        return []\n\n    def _dict_arg(self, name, value):\n        command = []\n        if value and isinstance(value, dict):\n            for prop, value in value.items():\n                command += [name, '{0}={1}'.format(prop, value)]\n        return command\n\n    def _flag_arg(self, name, value):\n        if value:\n            return [name]\n        return []\n\n\nclass PySparkTask(SparkSubmitTask):\n    \"\"\"\n    Template task for running an inline PySpark job\n\n    Simply implement the ``main`` method in your subclass\n\n    You can optionally define package names to be distributed to the cluster\n    with ``py_packages`` (uses luigi's global py-packages configuration by default)\n\n    \"\"\"\n\n    # Path to the pyspark program passed to spark-submit\n    app = os.path.join(os.path.dirname(__file__), 'pyspark_runner.py')\n\n    @property\n    def name(self):\n        return self.__class__.__name__\n\n    @property\n    def py_packages(self):\n        packages = configuration.get_config().get('spark', 'py-packages', None)\n        if packages:\n            return map(lambda s: s.strip(), packages.split(','))\n\n    @property\n    def files(self):\n        if self.deploy_mode == \"cluster\":\n            return [self.run_pickle]\n\n    def setup(self, conf):\n        \"\"\"\n        Called by the pyspark_runner with a SparkConf instance that will be used to instantiate the SparkContext\n\n        :param conf: SparkConf\n        \"\"\"\n\n    def setup_remote(self, sc):\n        self._setup_packages(sc)\n\n    def main(self, sc, *args):\n        \"\"\"\n        Called by the pyspark_runner with a SparkContext and any arguments returned by ``app_options()``\n\n        :param sc: SparkContext\n        :param args: arguments list\n        \"\"\"\n        raise NotImplementedError(\"subclass should define a main method\")\n\n    def app_command(self):\n        if self.deploy_mode == \"cluster\":\n            pickle_loc = os.path.basename(self.run_pickle)\n        else:\n            pickle_loc = self.run_pickle\n        return [self.app, pickle_loc] + self.app_options()\n\n    def run(self):\n        self.run_path = tempfile.mkdtemp(prefix=self.name)\n        self.run_pickle = os.path.join(self.run_path, '.'.join([self.name.replace(' ', '_'), 'pickle']))\n        with open(self.run_pickle, 'wb') as fd:\n            # Copy module file to run path.\n            module_path = os.path.abspath(inspect.getfile(self.__class__))\n            shutil.copy(module_path, os.path.join(self.run_path, '.'))\n            self._dump(fd)\n        try:\n            super(PySparkTask, self).run()\n        finally:\n            shutil.rmtree(self.run_path)\n\n    def _dump(self, fd):\n        with self.no_unpicklable_properties():\n            if self.__module__ == '__main__':\n                d = pickle.dumps(self)\n                module_name = os.path.basename(sys.argv[0]).rsplit('.', 1)[0]\n                d = d.replace(b'c__main__', b'c' + module_name.encode('ascii'))\n                fd.write(d)\n            else:\n                pickle.dump(self, fd)\n\n    def _setup_packages(self, sc):\n        \"\"\"\n        This method compresses and uploads packages to the cluster\n\n        \"\"\"\n        packages = self.py_packages\n        if not packages:\n            return\n        for package in packages:\n            mod = importlib.import_module(package)\n            try:\n                mod_path = mod.__path__[0]\n            except AttributeError:\n                mod_path = mod.__file__\n            tar_path = os.path.join(self.run_path, package + '.tar.gz')\n            tar = tarfile.open(tar_path, \"w:gz\")\n            tar.add(mod_path, os.path.basename(mod_path))\n            tar.close()\n            sc.addPyFile(tar_path)\n", "path": "luigi/contrib/spark.py"}], "after_files": [{"content": "# -*- coding: utf-8 -*-\n#\n# Copyright 2012-2015 Spotify AB\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n#\n\nimport logging\nimport os\nimport re\nimport sys\nimport tempfile\nimport shutil\nimport importlib\nimport tarfile\nimport inspect\ntry:\n    import cPickle as pickle\nexcept ImportError:\n    import pickle\n\nfrom luigi import six\nfrom luigi.contrib.external_program import ExternalProgramTask\nfrom luigi import configuration\n\nlogger = logging.getLogger('luigi-interface')\n\n\nclass SparkSubmitTask(ExternalProgramTask):\n    \"\"\"\n    Template task for running a Spark job\n\n    Supports running jobs on Spark local, standalone, Mesos or Yarn\n\n    See http://spark.apache.org/docs/latest/submitting-applications.html\n    for more information\n\n    \"\"\"\n\n    # Application (.jar or .py file)\n    name = None\n    entry_class = None\n    app = None\n\n    # Only log stderr if spark fails (since stderr is normally quite verbose)\n    always_log_stderr = False\n    # Spark applications write its logs into stderr\n    stream_for_searching_tracking_url = 'stderr'\n\n    def run(self):\n        if self.deploy_mode == \"cluster\":\n            # in cluster mode client only receives application status once a period of time\n            self.tracking_url_pattern = r\"tracking URL: (https?://.*)\\s\"\n        else:\n            self.tracking_url_pattern = r\"Bound (?:.*) to (?:.*), and started at (https?://.*)\\s\"\n\n        super(SparkSubmitTask, self).run()\n\n    def app_options(self):\n        \"\"\"\n        Subclass this method to map your task parameters to the app's arguments\n\n        \"\"\"\n        return []\n\n    @property\n    def spark_version(self):\n        return \"spark\"\n\n    @property\n    def spark_submit(self):\n        return configuration.get_config().get(self.spark_version, 'spark-submit', 'spark-submit')\n\n    @property\n    def master(self):\n        return configuration.get_config().get(self.spark_version, \"master\", None)\n\n    @property\n    def deploy_mode(self):\n        return configuration.get_config().get(self.spark_version, \"deploy-mode\", None)\n\n    @property\n    def jars(self):\n        return self._list_config(configuration.get_config().get(self.spark_version, \"jars\", None))\n\n    @property\n    def packages(self):\n        return self._list_config(configuration.get_config().get(\n            self.spark_version, \"packages\", None))\n\n    @property\n    def py_files(self):\n        return self._list_config(configuration.get_config().get(\n            self.spark_version, \"py-files\", None))\n\n    @property\n    def files(self):\n        return self._list_config(configuration.get_config().get(self.spark_version, \"files\", None))\n\n    @property\n    def conf(self):\n        return self._dict_config(configuration.get_config().get(self.spark_version, \"conf\", None))\n\n    @property\n    def properties_file(self):\n        return configuration.get_config().get(self.spark_version, \"properties-file\", None)\n\n    @property\n    def driver_memory(self):\n        return configuration.get_config().get(self.spark_version, \"driver-memory\", None)\n\n    @property\n    def driver_java_options(self):\n        return configuration.get_config().get(self.spark_version, \"driver-java-options\", None)\n\n    @property\n    def driver_library_path(self):\n        return configuration.get_config().get(self.spark_version, \"driver-library-path\", None)\n\n    @property\n    def driver_class_path(self):\n        return configuration.get_config().get(self.spark_version, \"driver-class-path\", None)\n\n    @property\n    def executor_memory(self):\n        return configuration.get_config().get(self.spark_version, \"executor-memory\", None)\n\n    @property\n    def driver_cores(self):\n        return configuration.get_config().get(self.spark_version, \"driver-cores\", None)\n\n    @property\n    def supervise(self):\n        return bool(configuration.get_config().get(self.spark_version, \"supervise\", False))\n\n    @property\n    def total_executor_cores(self):\n        return configuration.get_config().get(self.spark_version, \"total-executor-cores\", None)\n\n    @property\n    def executor_cores(self):\n        return configuration.get_config().get(self.spark_version, \"executor-cores\", None)\n\n    @property\n    def queue(self):\n        return configuration.get_config().get(self.spark_version, \"queue\", None)\n\n    @property\n    def num_executors(self):\n        return configuration.get_config().get(self.spark_version, \"num-executors\", None)\n\n    @property\n    def archives(self):\n        return self._list_config(configuration.get_config().get(\n            self.spark_version, \"archives\", None))\n\n    @property\n    def hadoop_conf_dir(self):\n        return configuration.get_config().get(self.spark_version, \"hadoop-conf-dir\", None)\n\n    def get_environment(self):\n        env = os.environ.copy()\n        hadoop_conf_dir = self.hadoop_conf_dir\n        if hadoop_conf_dir:\n            env['HADOOP_CONF_DIR'] = hadoop_conf_dir\n        return env\n\n    def program_environment(self):\n        return self.get_environment()\n\n    def program_args(self):\n        return self.spark_command() + self.app_command()\n\n    def spark_command(self):\n        command = [self.spark_submit]\n        command += self._text_arg('--master', self.master)\n        command += self._text_arg('--deploy-mode', self.deploy_mode)\n        command += self._text_arg('--name', self.name)\n        command += self._text_arg('--class', self.entry_class)\n        command += self._list_arg('--jars', self.jars)\n        command += self._list_arg('--packages', self.packages)\n        command += self._list_arg('--py-files', self.py_files)\n        command += self._list_arg('--files', self.files)\n        command += self._list_arg('--archives', self.archives)\n        command += self._dict_arg('--conf', self.conf)\n        command += self._text_arg('--properties-file', self.properties_file)\n        command += self._text_arg('--driver-memory', self.driver_memory)\n        command += self._text_arg('--driver-java-options', self.driver_java_options)\n        command += self._text_arg('--driver-library-path', self.driver_library_path)\n        command += self._text_arg('--driver-class-path', self.driver_class_path)\n        command += self._text_arg('--executor-memory', self.executor_memory)\n        command += self._text_arg('--driver-cores', self.driver_cores)\n        command += self._flag_arg('--supervise', self.supervise)\n        command += self._text_arg('--total-executor-cores', self.total_executor_cores)\n        command += self._text_arg('--executor-cores', self.executor_cores)\n        command += self._text_arg('--queue', self.queue)\n        command += self._text_arg('--num-executors', self.num_executors)\n        return command\n\n    def app_command(self):\n        if not self.app:\n            raise NotImplementedError(\"subclass should define an app (.jar or .py file)\")\n        return [self.app] + self.app_options()\n\n    def _list_config(self, config):\n        if config and isinstance(config, six.string_types):\n            return list(map(lambda x: x.strip(), config.split(',')))\n\n    def _dict_config(self, config):\n        if config and isinstance(config, six.string_types):\n            return dict(map(lambda i: i.split('=', 1), config.split('|')))\n\n    def _text_arg(self, name, value):\n        if value:\n            return [name, value]\n        return []\n\n    def _list_arg(self, name, value):\n        if value and isinstance(value, (list, tuple)):\n            return [name, ','.join(value)]\n        return []\n\n    def _dict_arg(self, name, value):\n        command = []\n        if value and isinstance(value, dict):\n            for prop, value in value.items():\n                command += [name, '{0}={1}'.format(prop, value)]\n        return command\n\n    def _flag_arg(self, name, value):\n        if value:\n            return [name]\n        return []\n\n\nclass PySparkTask(SparkSubmitTask):\n    \"\"\"\n    Template task for running an inline PySpark job\n\n    Simply implement the ``main`` method in your subclass\n\n    You can optionally define package names to be distributed to the cluster\n    with ``py_packages`` (uses luigi's global py-packages configuration by default)\n\n    \"\"\"\n\n    # Path to the pyspark program passed to spark-submit\n    app = os.path.join(os.path.dirname(__file__), 'pyspark_runner.py')\n\n    @property\n    def name(self):\n        return self.__class__.__name__\n\n    @property\n    def py_packages(self):\n        packages = configuration.get_config().get('spark', 'py-packages', None)\n        if packages:\n            return map(lambda s: s.strip(), packages.split(','))\n\n    @property\n    def files(self):\n        if self.deploy_mode == \"cluster\":\n            return [self.run_pickle]\n\n    def setup(self, conf):\n        \"\"\"\n        Called by the pyspark_runner with a SparkConf instance that will be used to instantiate the SparkContext\n\n        :param conf: SparkConf\n        \"\"\"\n\n    def setup_remote(self, sc):\n        self._setup_packages(sc)\n\n    def main(self, sc, *args):\n        \"\"\"\n        Called by the pyspark_runner with a SparkContext and any arguments returned by ``app_options()``\n\n        :param sc: SparkContext\n        :param args: arguments list\n        \"\"\"\n        raise NotImplementedError(\"subclass should define a main method\")\n\n    def app_command(self):\n        if self.deploy_mode == \"cluster\":\n            pickle_loc = os.path.basename(self.run_pickle)\n        else:\n            pickle_loc = self.run_pickle\n        return [self.app, pickle_loc] + self.app_options()\n\n    def run(self):\n        path_name_fragment = re.sub(r'[^\\w]', '_', self.name)\n        self.run_path = tempfile.mkdtemp(prefix=path_name_fragment)\n        self.run_pickle = os.path.join(self.run_path, '.'.join([path_name_fragment, 'pickle']))\n        with open(self.run_pickle, 'wb') as fd:\n            # Copy module file to run path.\n            module_path = os.path.abspath(inspect.getfile(self.__class__))\n            shutil.copy(module_path, os.path.join(self.run_path, '.'))\n            self._dump(fd)\n        try:\n            super(PySparkTask, self).run()\n        finally:\n            shutil.rmtree(self.run_path)\n\n    def _dump(self, fd):\n        with self.no_unpicklable_properties():\n            if self.__module__ == '__main__':\n                d = pickle.dumps(self)\n                module_name = os.path.basename(sys.argv[0]).rsplit('.', 1)[0]\n                d = d.replace(b'c__main__', b'c' + module_name.encode('ascii'))\n                fd.write(d)\n            else:\n                pickle.dump(self, fd)\n\n    def _setup_packages(self, sc):\n        \"\"\"\n        This method compresses and uploads packages to the cluster\n\n        \"\"\"\n        packages = self.py_packages\n        if not packages:\n            return\n        for package in packages:\n            mod = importlib.import_module(package)\n            try:\n                mod_path = mod.__path__[0]\n            except AttributeError:\n                mod_path = mod.__file__\n            tar_path = os.path.join(self.run_path, package + '.tar.gz')\n            tar = tarfile.open(tar_path, \"w:gz\")\n            tar.add(mod_path, os.path.basename(mod_path))\n            tar.close()\n            sc.addPyFile(tar_path)\n", "path": "luigi/contrib/spark.py"}]}