diff --git a/sdk/ml/azure-ai-ml/CHANGELOG.md b/sdk/ml/azure-ai-ml/CHANGELOG.md index 1d652082bbee..0aba4fdba0be 100644 --- a/sdk/ml/azure-ai-ml/CHANGELOG.md +++ b/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. diff --git a/sdk/ml/azure-ai-ml/api.md b/sdk/ml/azure-ai-ml/api.md index 8e629a502a9f..665b65977e7b 100644 --- a/sdk/ml/azure-ai-ml/api.md +++ b/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 diff --git a/sdk/ml/azure-ai-ml/api.metadata.yml b/sdk/ml/azure-ai-ml/api.metadata.yml index 1ddd7ebc1983..22811cf31ac7 100644 --- a/sdk/ml/azure-ai-ml/api.metadata.yml +++ b/sdk/ml/azure-ai-ml/api.metadata.yml @@ -1,3 +1,3 @@ -apiMdSha256: 4af4d65ad5c624b35aefa8442a07cb8cb147fe1dcbab6aa993625728a4a04de1 +apiMdSha256: 6244eccbf1e969c65201e0f0faa0f67261193b4145532ba886effecc54a8858a parserVersion: 0.3.30 pythonVersion: 3.12.13 diff --git a/sdk/ml/azure-ai-ml/azure/ai/ml/constants/_common.py b/sdk/ml/azure-ai-ml/azure/ai/ml/constants/_common.py index 2e12d7f41207..55df8f3823e8 100644 --- a/sdk/ml/azure-ai-ml/azure/ai/ml/constants/_common.py +++ b/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: diff --git a/sdk/ml/azure-ai-ml/azure/ai/ml/entities/_job/_input_output_helpers.py b/sdk/ml/azure-ai-ml/azure/ai/ml/entities/_job/_input_output_helpers.py index 41f617766466..651402d4dc3b 100644 --- a/sdk/ml/azure-ai-ml/azure/ai/ml/entities/_job/_input_output_helpers.py +++ b/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 diff --git a/sdk/ml/azure-ai-ml/azure/ai/ml/entities/_job/pipeline/_io/mixin.py b/sdk/ml/azure-ai-ml/azure/ai/ml/entities/_job/pipeline/_io/mixin.py index 1a22b9814016..a532ea8489e8 100644 --- a/sdk/ml/azure-ai-ml/azure/ai/ml/entities/_job/pipeline/_io/mixin.py +++ b/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]: diff --git a/sdk/ml/azure-ai-ml/azure/ai/ml/entities/_job/spark_helpers.py b/sdk/ml/azure-ai-ml/azure/ai/ml/entities/_job/spark_helpers.py index d3fdf9dc5fed..db9227d26117 100644 --- a/sdk/ml/azure-ai-ml/azure/ai/ml/entities/_job/spark_helpers.py +++ b/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,9 +193,9 @@ 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. @@ -199,12 +203,12 @@ def _validate_input_output_mode(inputs: Any, outputs: Any) -> None: # 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, ) diff --git a/sdk/ml/azure-ai-ml/tests/pipeline_job/unittests/test_pipeline_job_entity.py b/sdk/ml/azure-ai-ml/tests/pipeline_job/unittests/test_pipeline_job_entity.py index 52dbb263dc67..90fe9e6b28b2 100644 --- a/sdk/ml/azure-ai-ml/tests/pipeline_job/unittests/test_pipeline_job_entity.py +++ b/sdk/ml/azure-ai-ml/tests/pipeline_job/unittests/test_pipeline_job_entity.py @@ -1754,7 +1754,7 @@ 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", ), @@ -1762,7 +1762,7 @@ def test_spark_node_with_remote_component_in_pipeline( "./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", ), @@ -1770,7 +1770,7 @@ def test_spark_node_with_remote_component_in_pipeline( "./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", ), @@ -1778,7 +1778,7 @@ def test_spark_node_with_remote_component_in_pipeline( "./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", ), diff --git a/sdk/ml/azure-ai-ml/tests/pipeline_job/unittests/test_pipeline_job_schema.py b/sdk/ml/azure-ai-ml/tests/pipeline_job/unittests/test_pipeline_job_schema.py index e01dc2febed9..72d38ae85b23 100644 --- a/sdk/ml/azure-ai-ml/tests/pipeline_job/unittests/test_pipeline_job_schema.py +++ b/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: