[azure-ai-ml] Enable hdfs input/output mode for Spark jobs and components

sdk/ml/azure-ai-ml/CHANGELOG.md

@@ -3,6 +3,7 @@
 ## 1.35.0 (unreleased)
 
 ### Features Added
+- Added support for the `hdfs` input/output delivery mode (`InputOutputModes.HDFS`) for Spark jobs and components. Spark inputs and outputs can now be set to `hdfs` mode in addition to `direct`, allowing pipelines to explicitly request HDFS delivery instead of relying on service-side defaults.
 
 ### Bugs Fixed
 - Fixed internal pipeline `Command` node dropping node-level interactive `services` (SSH, JupyterLab, TensorBoard, VS Code, etc.) during serialization, which prevented interactive endpoints from being created for Singularity jobs. The `services` are now serialized into the pipeline REST request and round-tripped on deserialization, matching the public `Command` node behavior.

sdk/ml/azure-ai-ml/api.md

@@ -2485,6 +2485,7 @@ namespace azure.ai.ml.constants
         DOWNLOAD = download
         EVAL_DOWNLOAD = eval_download
         EVAL_MOUNT = eval_mount
+        HDFS = hdfs
         MOUNT = mount
         RO_MOUNT = ro_mount
         RW_MOUNT = rw_mount

sdk/ml/azure-ai-ml/api.metadata.yml

@@ -1,3 +1,3 @@
-apiMdSha256: 4af4d65ad5c624b35aefa8442a07cb8cb147fe1dcbab6aa993625728a4a04de1
+apiMdSha256: e66101de4aa1794cfa84a79fb27d7a39eed7afaeb714bf620d9a2251c15f8529
 parserVersion: 0.3.30
 pythonVersion: 3.12.13

sdk/ml/azure-ai-ml/azure/ai/ml/constants/_common.py

@@ -836,6 +836,8 @@ class InputOutputModes:
     """Evaluation download asset type."""
     DIRECT = "direct"
     """Direct asset type."""
+    HDFS = "hdfs"
+    """HDFS asset type."""
 
 
 class ConnectionTypes:

sdk/ml/azure-ai-ml/azure/ai/ml/entities/_job/_input_output_helpers.py

@@ -103,6 +103,8 @@ def to_hybrid_rest_model(value: Any, arm_base_cls: Any) -> Any:
     InputDeliveryMode.DIRECT: InputOutputModes.DIRECT,
     InputDeliveryMode.EVAL_MOUNT: InputOutputModes.EVAL_MOUNT,
     InputDeliveryMode.EVAL_DOWNLOAD: InputOutputModes.EVAL_DOWNLOAD,
+    "Hdfs": InputOutputModes.HDFS,
+    "hdfs": InputOutputModes.HDFS,
 }
 
 INPUT_MOUNT_MAPPING_TO_REST = {
@@ -113,6 +115,7 @@ def to_hybrid_rest_model(value: Any, arm_base_cls: Any) -> Any:
     InputOutputModes.EVAL_MOUNT: InputDeliveryMode.EVAL_MOUNT,
     InputOutputModes.EVAL_DOWNLOAD: InputDeliveryMode.EVAL_DOWNLOAD,
     InputOutputModes.DIRECT: InputDeliveryMode.DIRECT,
+    InputOutputModes.HDFS: "Hdfs",
 }
 
 
@@ -286,7 +289,7 @@ def to_rest_dataset_literal_inputs(
                     # reads ``val.mode`` back off the object, so keep them as plain (non-serialized)
                     # attributes on the arm_ml_service hybrid to preserve that behavior.
                     if input_value.mode:
-                        input_data.mode = INPUT_MOUNT_MAPPING_TO_REST[input_value.mode]
+                        input_data.mode = INPUT_MOUNT_MAPPING_TO_REST[input_value.mode.lower()]
                     if getattr(input_value, "path_on_compute", None) is not None:
                         input_data.pathOnCompute = input_value.path_on_compute
                     input_data.job_input_type = JobInputType.LITERAL

sdk/ml/azure-ai-ml/azure/ai/ml/entities/_job/pipeline/_io/mixin.py

@@ -270,7 +270,8 @@ def _input_entity_to_rest_inputs(cls, input_entity: Dict[str, Input]) -> Dict[st
         for name, val in rest_inputs.items():
             rest_dataset_literal_inputs[name] = _rest_io_to_snake_dict(val)
             if hasattr(val, "mode") and val.mode:
-                rest_dataset_literal_inputs[name].update({"mode": val.mode.value})
+                mode_val = val.mode.value if hasattr(val.mode, "value") else val.mode
+                rest_dataset_literal_inputs[name].update({"mode": mode_val})
         return rest_dataset_literal_inputs
 
     def _to_rest_outputs(self) -> Dict[str, Dict]:

sdk/ml/azure-ai-ml/azure/ai/ml/entities/_job/spark_helpers.py

@@ -121,16 +121,20 @@ def _validate_compute_or_resources(compute: Any, resources: Any) -> None:
         )
 
 
-# Only "direct" mode is supported for spark job inputs and outputs
+# Only "direct" and "hdfs" modes are supported for spark job inputs and outputs
+_SPARK_SUPPORTED_MODES = (InputOutputModes.DIRECT, InputOutputModes.HDFS)
+_SPARK_SUPPORTED_MODES_MSG = " and ".join("'{}'".format(mode) for mode in _SPARK_SUPPORTED_MODES)
+
+
 # pylint: disable=no-else-raise, too-many-boolean-expressions
 def _validate_input_output_mode(inputs: Any, outputs: Any) -> None:
     for input_name, input_value in inputs.items():
-        if isinstance(input_value, Input) and input_value.mode != InputOutputModes.DIRECT:
+        if isinstance(input_value, Input) and input_value.mode not in _SPARK_SUPPORTED_MODES:
             # For standalone job input
-            msg = "Input '{}' is using '{}' mode, only '{}' is supported for Spark job"
+            msg = "Input '{}' is using '{}' mode, only {} are supported for Spark job"
             raise ValidationException(
-                message=msg.format(input_name, input_value.mode, InputOutputModes.DIRECT),
-                no_personal_data_message=msg.format("[input_name]", "[input_value.mode]", "direct"),
+                message=msg.format(input_name, input_value.mode, _SPARK_SUPPORTED_MODES_MSG),
+                no_personal_data_message=msg.format("[input_name]", "[input_value.mode]", _SPARK_SUPPORTED_MODES_MSG),
                 target=ErrorTarget.SPARK_JOB,
                 error_category=ErrorCategory.USER_ERROR,
             )
@@ -142,9 +146,9 @@ def _validate_input_output_mode(inputs: Any, outputs: Any) -> None:
                     isinstance(input_value._data.path, str)
                     and bool(re.search(ComponentJobConstants.INPUT_PATTERN, input_value._data.path))
                 )
-                and input_value._data.mode != InputOutputModes.DIRECT
+                and input_value._data.mode not in _SPARK_SUPPORTED_MODES
             )
-            and (isinstance(input_value._meta, Input) and input_value._meta.mode != InputOutputModes.DIRECT)
+            and (isinstance(input_value._meta, Input) and input_value._meta.mode not in _SPARK_SUPPORTED_MODES)
         ):
             # For node input in pipeline job, client side can only validate node input which isn't bound to pipeline
             # input or node output.
@@ -156,12 +160,12 @@ def _validate_input_output_mode(inputs: Any, outputs: Any) -> None:
             # always get None mode in node level. In this case, if we define correct "Direct" mode in component yaml,
             # component level mode will take effect and run successfully. Otherwise, it need to set mode in node level
             # like input1: path: ${{parent.jobs.sample_word.outputs.output1}} mode: direct.
-            msg = "Input '{}' is using '{}' mode, only '{}' is supported for Spark job"
+            msg = "Input '{}' is using '{}' mode, only {} are supported for Spark job"
             raise ValidationException(
                 message=msg.format(
-                    input_name, input_value._data.mode or input_value._meta.mode, InputOutputModes.DIRECT
+                    input_name, input_value._data.mode or input_value._meta.mode, _SPARK_SUPPORTED_MODES_MSG
                 ),
-                no_personal_data_message=msg.format("[input_name]", "[input_value.mode]", "direct"),
+                no_personal_data_message=msg.format("[input_name]", "[input_value.mode]", _SPARK_SUPPORTED_MODES_MSG),
                 target=ErrorTarget.SPARK_JOB,
                 error_category=ErrorCategory.USER_ERROR,
             )
@@ -170,13 +174,13 @@ def _validate_input_output_mode(inputs: Any, outputs: Any) -> None:
         if (
             isinstance(output_value, Output)
             and output_name != "default"
-            and output_value.mode != InputOutputModes.DIRECT
+            and output_value.mode not in _SPARK_SUPPORTED_MODES
         ):
             # For standalone job output
-            msg = "Output '{}' is using '{}' mode, only '{}' is supported for Spark job"
+            msg = "Output '{}' is using '{}' mode, only {} are supported for Spark job"
             raise ValidationException(
-                message=msg.format(output_name, output_value.mode, InputOutputModes.DIRECT),
-                no_personal_data_message=msg.format("[output_name]", "[output_value.mode]", "direct"),
+                message=msg.format(output_name, output_value.mode, _SPARK_SUPPORTED_MODES_MSG),
+                no_personal_data_message=msg.format("[output_name]", "[output_value.mode]", _SPARK_SUPPORTED_MODES_MSG),
                 target=ErrorTarget.SPARK_JOB,
                 error_category=ErrorCategory.USER_ERROR,
             )
@@ -189,22 +193,22 @@ def _validate_input_output_mode(inputs: Any, outputs: Any) -> None:
                     isinstance(output_value._data.path, str)
                     and bool(re.search(ComponentJobConstants.OUTPUT_PATTERN, output_value._data.path))
                 )
-                and output_value._data.mode != InputOutputModes.DIRECT
+                and output_value._data.mode not in _SPARK_SUPPORTED_MODES
             )
-            and (isinstance(output_value._meta, Output) and output_value._meta.mode != InputOutputModes.DIRECT)
+            and (isinstance(output_value._meta, Output) and output_value._meta.mode not in _SPARK_SUPPORTED_MODES)
         ):
             # For node output in pipeline job, client side can only validate node output which isn't bound to pipeline
             # output.
             # 1. If node output is bound to pipeline output, we can't get pipeline level output mode in node level
             # validate. Even if we can judge through component output mode (_meta), we should note that pipeline level
             # output mode has higher priority than component level. so component output can be set "upload", but it
             # can run successfully when pipeline output is "Direct".
-            msg = "Output '{}' is using '{}' mode, only '{}' is supported for Spark job"
+            msg = "Output '{}' is using '{}' mode, only {} are supported for Spark job"
             raise ValidationException(
                 message=msg.format(
-                    output_name, output_value._data.mode or output_value._meta.mode, InputOutputModes.DIRECT
+                    output_name, output_value._data.mode or output_value._meta.mode, _SPARK_SUPPORTED_MODES_MSG
                 ),
-                no_personal_data_message=msg.format("[output_name]", "[output_value.mode]", "direct"),
+                no_personal_data_message=msg.format("[output_name]", "[output_value.mode]", _SPARK_SUPPORTED_MODES_MSG),
                 target=ErrorTarget.SPARK_JOB,
                 error_category=ErrorCategory.USER_ERROR,
             )

sdk/ml/azure-ai-ml/tests/pipeline_job/unittests/test_pipeline_job_entity.py

@@ -1754,31 +1754,31 @@ def test_spark_node_with_remote_component_in_pipeline(
                 "./tests/test_configs/pipeline_jobs/invalid/pipeline_job_with_spark_job_with_invalid_output_mode.yml",
                 {
                     "jobs.hello_world.outputs.output": "Output 'output' is using 'None' mode, "
-                    "only 'direct' is supported for Spark job"
+                    "only 'direct' and 'hdfs' are supported for Spark job"
                 },
                 id="none_mode_output",
             ),
             pytest.param(
                 "./tests/test_configs/pipeline_jobs/invalid/pipeline_job_with_spark_job_with_invalid_component_output_mode.yml",
                 {
                     "jobs.hello_world.outputs.output1": "Output 'output1' is using 'upload' mode, "
-                    "only 'direct' is supported for Spark job"
+                    "only 'direct' and 'hdfs' are supported for Spark job"
                 },
                 id="upload_mode_output",
             ),
             pytest.param(
                 "./tests/test_configs/pipeline_jobs/invalid/pipeline_job_with_spark_job_with_invalid_input_mode.yml",
                 {
                     "jobs.hello_world.inputs.file_input1": "Input 'file_input1' is using 'None' mode, "
-                    "only 'direct' is supported for Spark job",
+                    "only 'direct' and 'hdfs' are supported for Spark job",
                 },
                 id="none_mode_input",
             ),
             pytest.param(
                 "./tests/test_configs/pipeline_jobs/invalid/pipeline_job_with_spark_job_with_invalid_component_input_mode.yml",
                 {
                     "jobs.hello_world.inputs.input1": "Input 'input1' is using 'mount' mode, "
-                    "only 'direct' is supported for Spark job"
+                    "only 'direct' and 'hdfs' are supported for Spark job"
                 },
                 id="mount_mode_input",
             ),

sdk/ml/azure-ai-ml/tests/pipeline_job/unittests/test_pipeline_job_schema.py

@@ -48,6 +48,8 @@
 from azure.ai.ml.entities._inputs_outputs import Input, Output
 from azure.ai.ml.entities._job._input_output_helpers import (
     INPUT_MOUNT_MAPPING_FROM_REST,
+    from_rest_inputs_to_dataset_literal,
+    to_rest_dataset_literal_inputs,
     validate_pipeline_input_key_characters,
 )
 from azure.ai.ml.entities._job.automl.search_space_utils import (
@@ -71,6 +73,49 @@
 @pytest.mark.unittest
 @pytest.mark.pipeline_test
 class TestPipelineJobSchema:
+    def test_hdfs_input_mode_round_trip(self):
+        rest_inputs = to_rest_dataset_literal_inputs(
+            {
+                "hdfs_input": Input(
+                    type=AssetTypes.URI_FOLDER,
+                    path="azureml://datastores/workspaceblobstore/paths/data/",
+                    mode=InputOutputModes.HDFS,
+                )
+            },
+            job_type=None,
+        )
+
+        assert rest_inputs["hdfs_input"].mode == "Hdfs"
+
+        sdk_inputs = from_rest_inputs_to_dataset_literal(rest_inputs)
+        assert isinstance(sdk_inputs["hdfs_input"], Input)
+        assert sdk_inputs["hdfs_input"].mode == InputOutputModes.HDFS
+
+    def test_spark_input_output_hdfs_mode_supported(self):
+        from azure.ai.ml.entities._job.spark_helpers import _validate_input_output_mode
+
+        # "hdfs" mode must be accepted for Spark inputs and outputs (ICM 586585073).
+        _validate_input_output_mode(
+            inputs={"hdfs_input": Input(type=AssetTypes.URI_FOLDER, path="test", mode=InputOutputModes.HDFS)},
+            outputs={"hdfs_output": Output(type=AssetTypes.URI_FOLDER, mode=InputOutputModes.HDFS)},
+        )
+        # "direct" mode remains supported.
+        _validate_input_output_mode(
+            inputs={"direct_input": Input(type=AssetTypes.URI_FOLDER, path="test", mode=InputOutputModes.DIRECT)},
+            outputs={},
+        )
+
+    def test_spark_input_unsupported_mode_rejected(self):
+        from azure.ai.ml.entities._job.spark_helpers import _validate_input_output_mode
+        from azure.ai.ml.exceptions import ValidationException
+
+        with pytest.raises(ValidationException) as exc_info:
+            _validate_input_output_mode(
+                inputs={"bad_input": Input(type=AssetTypes.URI_FOLDER, path="test", mode=InputOutputModes.MOUNT)},
+                outputs={},
+            )
+        assert "only 'direct' and 'hdfs' are supported for Spark job" in str(exc_info.value)
+
     def test_validate_pipeline_job_keys(self):
         def validator(key, assert_valid=True):
             if assert_valid: