feat(feature-processor): Wire image resolver into pipeline and unlock dependencies

Replace _get_default_spark_image() call in feature_scheduler.py with
_get_spark_image_uri() from the new image resolver module. Remove
pyspark==3.3.2 version pin and switch to unified
sagemaker-feature-store-pyspark connector package in pyproject.toml.

---
X-AI-Prompt: wire image resolver into feature_scheduler.py and update pyproject.toml dependencies
X-AI-Tool: kiro

Author: BassemHalim

sagemaker-mlops/pyproject.toml

@@ -32,8 +32,7 @@ dependencies = [
 
 [project.optional-dependencies]
 feature-processor = [
-    "pyspark==3.3.2",
-    "sagemaker-feature-store-pyspark-3.3",
+    "sagemaker-feature-store-pyspark",
     "setuptools<82",
 ]
 
@@ -42,8 +41,7 @@ test = [
     "pytest-cov",
     "mock",
     "setuptools<82",
-    "pyspark==3.3.2",
-    "sagemaker-feature-store-pyspark-3.3",
+    "sagemaker-feature-store-pyspark",
     "pandas<3.0",
     "numpy<3.0",
 ]

sagemaker-mlops/src/sagemaker/mlops/feature_store/feature_processor/feature_scheduler.py

@@ -98,6 +98,7 @@
     FeatureProcessorLineageHandler,
     TransformationCode,
 )
+from sagemaker.mlops.feature_store.feature_processor._image_resolver import _get_spark_image_uri
 
 from sagemaker.core.remote_function.job import (
     _JobSettings,
@@ -1039,7 +1040,7 @@ def _get_remote_decorator_config_from_input(
     # TODO: This needs to be removed when new mode is introduced.
     if remote_decorator_config.spark_config is None:
         remote_decorator_config.spark_config = SparkConfig()
-    remote_decorator_config.image_uri = _JobSettings._get_default_spark_image(sagemaker_session)
+    remote_decorator_config.image_uri = _get_spark_image_uri(sagemaker_session)
 
     return remote_decorator_config
 

sagemaker-mlops/tests/unit/sagemaker/mlops/feature_store/feature_processor/test_feature_scheduler.py

@@ -162,6 +162,10 @@ def config_uploader():
     return uploader
 
 
+@patch(
+    "sagemaker.core.remote_function.job._JobSettings._get_default_spark_image",
+    new=Mock(return_value="some_image_uri"),
+)
 @patch(
     "sagemaker.mlops.feature_store.feature_processor.feature_scheduler._validate_fg_lineage_resources",
     return_value=None,
@@ -171,7 +175,7 @@ def config_uploader():
     return_value=mock_pipeline(),
 )
 @patch(
-    "sagemaker.core.remote_function.job._JobSettings._get_default_spark_image",
+    "sagemaker.mlops.feature_store.feature_processor.feature_scheduler._get_spark_image_uri",
     return_value="some_image_uri",
 )
 @patch("sagemaker.mlops.feature_store.feature_processor._config_uploader.TrainingInput")
@@ -470,9 +474,13 @@ def test_to_pipeline_not_wrapped_by_remote(get_execution_role, session):
         )
 
 
-@patch("sagemaker.core.remote_function.job.Session", return_value=mock_session())
 @patch(
     "sagemaker.core.remote_function.job._JobSettings._get_default_spark_image",
+    new=Mock(return_value="some_image_uri"),
+)
+@patch("sagemaker.core.remote_function.job.Session", return_value=mock_session())
+@patch(
+    "sagemaker.mlops.feature_store.feature_processor.feature_scheduler._get_spark_image_uri",
     return_value="some_image_uri",
 )
 @patch("sagemaker.core.remote_function.job.get_execution_role", return_value=EXECUTION_ROLE_ARN)
@@ -522,9 +530,13 @@ def test_to_pipeline_wrong_mode(get_execution_role, mock_spark_image, session):
         )
 
 
-@patch("sagemaker.core.remote_function.job.Session", return_value=mock_session())
 @patch(
     "sagemaker.core.remote_function.job._JobSettings._get_default_spark_image",
+    new=Mock(return_value="some_image_uri"),
+)
+@patch("sagemaker.core.remote_function.job.Session", return_value=mock_session())
+@patch(
+    "sagemaker.mlops.feature_store.feature_processor.feature_scheduler._get_spark_image_uri",
     return_value="some_image_uri",
 )
 @patch("sagemaker.core.remote_function.job.get_execution_role", return_value=EXECUTION_ROLE_ARN)
@@ -577,9 +589,13 @@ def test_to_pipeline_pipeline_name_length_limit_exceeds(
         )
 
 
-@patch("sagemaker.core.remote_function.job.Session", return_value=mock_session())
 @patch(
     "sagemaker.core.remote_function.job._JobSettings._get_default_spark_image",
+    new=Mock(return_value="some_image_uri"),
+)
+@patch("sagemaker.core.remote_function.job.Session", return_value=mock_session())
+@patch(
+    "sagemaker.mlops.feature_store.feature_processor.feature_scheduler._get_spark_image_uri",
     return_value="some_image_uri",
 )
 @patch("sagemaker.core.remote_function.job.get_execution_role", return_value=EXECUTION_ROLE_ARN)