feat(feature-processor): Add dynamic Hadoop version resolution

Replace hardcoded hadoop-aws:3.3.1 and hadoop-common:3.3.1 Maven
coordinates with dynamically resolved versions based on the installed
PySpark version. Add SPARK_TO_HADOOP_MAP supporting Spark 3.1-3.5
with fallback to latest known Hadoop version for unknown Spark versions.

---
X-AI-Prompt: implement dynamic hadoop version resolution in _spark_factory.py for multi-spark version compatibility
X-AI-Tool: kiro

Author: BassemHalim

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

@@ -13,10 +13,12 @@
 """Contains factory classes for instantiating Spark objects."""
 from __future__ import absolute_import
 
+import logging
 from functools import lru_cache
 from typing import List, Tuple, Dict
 
 import feature_store_pyspark
+import pyspark
 import feature_store_pyspark.FeatureStoreManager as fsm
 from pyspark.conf import SparkConf
 from pyspark.context import SparkContext
@@ -26,6 +28,33 @@
 
 SPARK_APP_NAME = "FeatureProcessor"
 
+logger = logging.getLogger(__name__)
+
+SPARK_TO_HADOOP_MAP = {
+    "3.1": "3.2.0",
+    "3.2": "3.3.1",
+    "3.3": "3.3.2",
+    "3.4": "3.3.4",
+    "3.5": "3.3.4",
+}
+
+_DEFAULT_HADOOP_VERSION = "3.3.4"
+
+
+def _get_hadoop_version():
+    """Resolve the Hadoop version for the installed PySpark version."""
+    spark_version = pyspark.__version__
+    major_minor = ".".join(spark_version.split(".")[:2])
+    hadoop_version = SPARK_TO_HADOOP_MAP.get(major_minor)
+    if hadoop_version is None:
+        hadoop_version = _DEFAULT_HADOOP_VERSION
+        logger.warning(
+            "Unknown Spark version %s. Falling back to Hadoop %s.",
+            spark_version,
+            hadoop_version,
+        )
+    return hadoop_version
+
 
 class SparkSessionFactory:
     """Lazy loading, memoizing, instantiation of SparkSessions.
@@ -116,9 +145,10 @@ def _get_spark_configs(self, is_training_job) -> List[Tuple[str, str]]:
 
         if not is_training_job:
             fp_spark_jars = feature_store_pyspark.classpath_jars()
+            hadoop_version = _get_hadoop_version()
             fp_spark_packages = [
-                "org.apache.hadoop:hadoop-aws:3.3.1",
-                "org.apache.hadoop:hadoop-common:3.3.1",
+                f"org.apache.hadoop:hadoop-aws:{hadoop_version}",
+                f"org.apache.hadoop:hadoop-common:{hadoop_version}",
             ]
 
             if self.spark_config and "spark.jars" in self.spark_config:

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

@@ -14,6 +14,7 @@
 from __future__ import absolute_import
 
 import feature_store_pyspark
+import pyspark
 import pytest
 from mock import Mock, patch, call
 
@@ -72,10 +73,12 @@ def test_spark_session_factory_configuration():
 
     # Verify configurations when not running on a training job
     assert ",".join(feature_store_pyspark.classpath_jars()) in spark_configs.get("spark.jars")
+    from sagemaker.mlops.feature_store.feature_processor._spark_factory import _get_hadoop_version
+    hadoop_version = _get_hadoop_version()
     assert ",".join(
         [
-            "org.apache.hadoop:hadoop-aws:3.3.1",
-            "org.apache.hadoop:hadoop-common:3.3.1",
+            f"org.apache.hadoop:hadoop-aws:{hadoop_version}",
+            f"org.apache.hadoop:hadoop-common:{hadoop_version}",
         ]
     ) in spark_configs.get("spark.jars.packages")
 
@@ -173,3 +176,34 @@ def test_spark_session_factory_get_spark_session_with_iceberg_config(env_helper)
         == "smfs.shaded.org.apache.iceberg.aws.s3.S3FileIO"
     )
     assert iceberg_configs.get("spark.sql.catalog.catalog.glue.skip-name-validation") == "true"
+
+
+@pytest.mark.parametrize(
+    "spark_version,expected_hadoop",
+    [
+        ("3.1.3", "3.2.0"),
+        ("3.2.2", "3.3.1"),
+        ("3.3.2", "3.3.2"),
+        ("3.4.1", "3.3.4"),
+        ("3.5.1", "3.3.4"),
+    ],
+)
+def test_get_hadoop_version(spark_version, expected_hadoop):
+    with patch.object(pyspark, "__version__", spark_version):
+        from sagemaker.mlops.feature_store.feature_processor._spark_factory import _get_hadoop_version
+        assert _get_hadoop_version() == expected_hadoop
+
+
+def test_get_hadoop_version_unknown_falls_back():
+    with patch.object(pyspark, "__version__", "3.6.0"):
+        from sagemaker.mlops.feature_store.feature_processor._spark_factory import _get_hadoop_version
+        assert _get_hadoop_version() == "3.3.4"
+
+
+def test_spark_configs_use_dynamic_hadoop_version():
+    with patch.object(pyspark, "__version__", "3.5.1"):
+        env_helper = Mock()
+        factory = SparkSessionFactory(env_helper)
+        configs = dict(factory._get_spark_configs(is_training_job=False))
+        assert "org.apache.hadoop:hadoop-aws:3.3.4" in configs.get("spark.jars.packages")
+        assert "org.apache.hadoop:hadoop-common:3.3.4" in configs.get("spark.jars.packages")