fix: install sagemaker-core in Spark container pre-execution

The Spark processing image does not have sagemaker-core pre-installed,
causing spark_app.py to fail with ModuleNotFoundError when importing
sagemaker.core.remote_function. Add sagemaker-core to the pre-execution
commands when spark_config is set.

Author: lucasjia-aws

sagemaker-core/src/sagemaker/core/remote_function/job.py

@@ -670,9 +670,23 @@ def __init__(
             sagemaker_session=self.sagemaker_session,
         )
 
-        # When using Spark, ensure sagemaker-feature-store-pyspark is installed
-        # and its version-matched JAR is on Spark's classpath before spark-submit
+        # When using Spark, ensure sagemaker-core and sagemaker-feature-store-pyspark
+        # are installed and the version-matched JAR is on Spark's classpath before spark-submit.
+        # The Spark processing image does not include sagemaker-core, so it must be installed
+        # explicitly for spark_app.py to import sagemaker.core.remote_function.
         if spark_config:
+            sagemaker_pysdk_version = (
+                RuntimeEnvironmentManager()._current_sagemaker_pysdk_version()
+            )
+            if sagemaker_pysdk_version and not sagemaker_pysdk_version.endswith(".dev0"):
+                install_sagemaker_cmd = (
+                    f"pip install --root-user-action=ignore"
+                    f" 'sagemaker-core=={sagemaker_pysdk_version}'"
+                )
+            else:
+                install_sagemaker_cmd = (
+                    "pip install --root-user-action=ignore 'sagemaker-core'"
+                )
             install_cmd = (
                 "pip install --root-user-action=ignore"
                 " 'sagemaker-feature-store-pyspark>=2,<3'"
@@ -690,10 +704,13 @@ def __init__(
                 "for j in glob.glob(os.path.join(jars_dir, '*' + spark_ver + '*.jar'))]\""
             )
             if self.pre_execution_commands is None:
-                self.pre_execution_commands = [install_cmd, copy_jar_cmd]
-            elif install_cmd not in self.pre_execution_commands:
-                self.pre_execution_commands.append(install_cmd)
-                self.pre_execution_commands.append(copy_jar_cmd)
+                self.pre_execution_commands = [install_sagemaker_cmd, install_cmd, copy_jar_cmd]
+            else:
+                if install_sagemaker_cmd not in self.pre_execution_commands:
+                    self.pre_execution_commands.insert(0, install_sagemaker_cmd)
+                if install_cmd not in self.pre_execution_commands:
+                    self.pre_execution_commands.append(install_cmd)
+                    self.pre_execution_commands.append(copy_jar_cmd)
 
         self.pre_execution_script = resolve_value_from_config(
             direct_input=pre_execution_script,