[dagster-aws] Honor configured EMR s3 job package path (#33733)

## Summary & Motivation

Closes #3679.

Patched the EMR step launcher so --py-files now matches the configured
distribution mode in python_modules/libraries/dagster-
aws/dagster_aws/emr/pyspark_step_launcher.py:436. If s3_job_package_path
is set, it uses that URI; if deploy_local_job_package=True, it uses the
staged code.zip; otherwise it omits --py-files entirely instead of
pointing at a non-uploaded artifact.

## Test Plan

## Changelog

- [dagster-aws] honor configured EMR s3 job package path

Author: cmpadden

python_modules/libraries/dagster-aws/dagster_aws/emr/pyspark_step_launcher.py

@@ -433,6 +433,12 @@ def _get_emr_step_def(self, run_id, step_key, solid_name):
             ).format(conf.get("spark.master")),
         )
 
+        py_files = []
+        if self.s3_job_package_path:
+            py_files = ["--py-files", self.s3_job_package_path]
+        elif self.deploy_local_job_package:
+            py_files = ["--py-files", self._artifact_s3_uri(run_id, step_key, CODE_ZIP_NAME)]
+
         command = (
             [
                 EMR_SPARK_HOME + "bin/spark-submit",
@@ -442,9 +448,8 @@ def _get_emr_step_def(self, run_id, step_key, solid_name):
                 conf.get("spark.submit.deployMode", "client"),
             ]
             + format_for_cli(list(flatten_dict(conf)))
+            + py_files
             + [
-                "--py-files",
-                self._artifact_s3_uri(run_id, step_key, CODE_ZIP_NAME),
                 self._artifact_s3_uri(run_id, step_key, self._main_file_name()),
                 self.staging_bucket,
                 self._artifact_s3_key(run_id, step_key, PICKLED_STEP_RUN_REF_FILE_NAME),

python_modules/libraries/dagster-aws/dagster_aws_tests/emr_tests/test_pyspark_step_launcher.py

@@ -112,3 +112,41 @@ def test_emr_pyspark_step_launcher_legacy_arguments():
             }
         )
     )
+
+
+def test_emr_step_def_uses_configured_s3_job_package_path():
+    launcher = EmrPySparkStepLauncher(
+        region_name="us-west-1",
+        staging_bucket="bucket",
+        staging_prefix="prefix",
+        wait_for_logs=False,
+        action_on_failure="CANCEL_AND_WAIT",
+        cluster_id="j-123",
+        spark_config={},
+        local_job_package_path=os.path.abspath(os.path.dirname(__file__)),
+        deploy_local_job_package=False,
+        s3_job_package_path="s3://configured/code.zip",
+    )
+
+    args = launcher._get_emr_step_def("run1", "step1", "op1")["HadoopJarStep"]["Args"]  # noqa: SLF001
+
+    assert "--py-files" in args
+    assert args[args.index("--py-files") + 1] == "s3://configured/code.zip"
+
+
+def test_emr_step_def_omits_py_files_when_code_is_preinstalled():
+    launcher = EmrPySparkStepLauncher(
+        region_name="us-west-1",
+        staging_bucket="bucket",
+        staging_prefix="prefix",
+        wait_for_logs=False,
+        action_on_failure="CANCEL_AND_WAIT",
+        cluster_id="j-123",
+        spark_config={},
+        local_job_package_path=os.path.abspath(os.path.dirname(__file__)),
+        deploy_local_job_package=False,
+    )
+
+    args = launcher._get_emr_step_def("run1", "step1", "op1")["HadoopJarStep"]["Args"]  # noqa: SLF001
+
+    assert "--py-files" not in args