PySparkTask: special characters in name (#2778)

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__))

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