Repository: spotify/luigi
Pull request: PySparkTask: handle special characters in name (#2778)
Fixes #2778 

## Description
It cleans _name_ property before it is used to derive _run_path_ and _run_pickle_ properties.

## Motivation and Context
If name property contains some special characters PySparkTask fails: #2778 

## Have you tested this? If so, how?
Unit test is included


Problem statement:
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.

Hints from issue discussion:

Commit message: PySparkTask: special characters in name (#2778)

Patch to review:
diff --git a/luigi/contrib/spark.py b/luigi/contrib/spark.py
index d84d08312c..bb9b0d400c 100644
--- 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 @@ def app_command(self):
         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']))
+        name = re.sub(r'[^\w]', '_', self.name)
+        self.run_path = tempfile.mkdtemp(prefix=name)
+        self.run_pickle = os.path.join(self.run_path, '.'.join([name, 'pickle']))
         with open(self.run_pickle, 'wb') as fd:
             # Copy module file to run path.
             module_path = os.path.abspath(inspect.getfile(self.__class__))
diff --git a/test/contrib/spark_test.py b/test/contrib/spark_test.py
index 74344d188a..45341051e3 100644
--- a/test/contrib/spark_test.py
+++ b/test/contrib/spark_test.py
@@ -97,6 +97,10 @@ def main(self, sc, *args):
         sc.textFile(self.input().path).saveAsTextFile(self.output().path)
 
 
+class MessyNamePySparkTask(TestPySparkTask):
+    name = 'AppName(a,b,c,1:2,3/4)'
+
+
 @attr('apache')
 class SparkSubmitTaskTest(unittest.TestCase):
     ss = 'ss-stub'
@@ -289,3 +293,10 @@ def mock_spark_submit(task):
 
         sc.textFile.assert_called_with('input')
         sc.textFile.return_value.saveAsTextFile.assert_called_with('output')
+
+    @patch('luigi.contrib.external_program.subprocess.Popen')
+    def test_name_cleanup(self, proc):
+        setup_run_process(proc)
+        job = MessyNamePySparkTask()
+        job.run()
+        assert 'AppName_a_b_c_1_2_3_4_' in job.run_path


Code review comments:
diff_hunk: @@ -298,8 +299,9 @@ def app_command(self):
         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']))
+        name = re.sub(r'[^\w]', '_', self.name)
path: luigi/contrib/spark.py
text: @user1:
I'd rename variable `name` to something like `run_path_prefix` (maybe some better option) due to two reasons: 
- it is actually a path, not a name anymore
- we must be having a hard time dealing with `name` and `self.name` in the same place

@author:
Good point! Renamed.