fix: fix failures for multiple spark run() invocations (#1979)

* reset command for each run invocation

* add unit test for command reset

* fix flake8

Co-authored-by: Ajay Karpur <[email protected]>

Author: guoqiao1992

src/sagemaker/spark/processing.py

@@ -737,6 +737,7 @@ def run(
                 user code file (default: None).
         """
         self._current_job_name = self._generate_current_job_name(job_name=job_name)
+        self.command = [_SparkProcessorBase._default_command]
 
         if not submit_app:
             raise ValueError("submit_app is required")
@@ -917,6 +918,7 @@ def run(
                 user code file (default: None).
         """
         self._current_job_name = self._generate_current_job_name(job_name=job_name)
+        self.command = [_SparkProcessorBase._default_command]
 
         if not submit_app:
             raise ValueError("submit_app is required")

tests/unit/sagemaker/spark/test_processing.py

@@ -738,6 +738,34 @@ def test_py_spark_processor_run(
         )
 
 
+@patch("sagemaker.spark.processing._SparkProcessorBase.run")
+@patch("sagemaker.spark.processing._SparkProcessorBase._stage_submit_deps")
+@patch("sagemaker.spark.processing._SparkProcessorBase._generate_current_job_name")
+def test_py_spark_processor_run_twice(
+    mock_generate_current_job_name, mock_stage_submit_deps, mock_super_run, py_spark_processor
+):
+    mock_stage_submit_deps.return_value = (processing_input, "opt")
+    mock_generate_current_job_name.return_value = "jobName"
+    expected_command = ["smspark-submit", "--py-files", "opt", "--jars", "opt", "--files", "opt"]
+    py_spark_processor.run(
+        submit_app="submit_app",
+        submit_py_files="files",
+        submit_jars="test",
+        submit_files="test",
+        inputs=[],
+    )
+
+    py_spark_processor.run(
+        submit_app="submit_app",
+        submit_py_files="files",
+        submit_jars="test",
+        submit_files="test",
+        inputs=[],
+    )
+
+    assert py_spark_processor.command == expected_command
+
+
 @pytest.mark.parametrize(
     "config, expected",
     [