Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions docs/spelling_wordlist.txt
Original file line number Diff line number Diff line change
Expand Up @@ -538,6 +538,7 @@ DisplayVideo
distcp
distro
distros
dkr
Dlp
dlp
DlpJob
Expand Down Expand Up @@ -2082,6 +2083,7 @@ XComArgs
xcomresult
XComs
Xero
xlarge
Xiaodong
xml
xmltodict
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -40,24 +40,43 @@ class SageMakerNotebookHook(BaseHook):
from airflow.providers.amazon.aws.hooks.sagemaker_unified_studio import SageMakerNotebookHook

notebook_hook = SageMakerNotebookHook(
execution_name="notebook_execution",
domain_id="dzd-example123456",
project_id="example123456",
input_config={"input_path": "path/to/notebook.ipynb", "input_params": {"param1": "value1"}},
output_config={"output_uri": "folder/output/location/prefix", "output_formats": "NOTEBOOK"},
execution_name="notebook_execution",
domain_region="us-east-1",
waiter_delay=10,
waiter_max_attempts=1440,
)

:param execution_name: The name of the notebook job to be executed, this is same as task_id.
:param domain_id: The domain ID for Amazon SageMaker Unified Studio. Optional - if not provided,
the SDK will attempt to resolve it from the MWAA environment.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is technically possible to export the right variables in a non-MWAA env, no?

Suggested change
the SDK will attempt to resolve it from the MWAA environment.
the SDK will attempt to resolve it from the environment.

Same suggestion for below and in the operator module.

:param project_id: The project ID for Amazon SageMaker Unified Studio. Optional - if not provided,
the SDK will attempt to resolve it from the MWAA environment.
:param input_config: Configuration for the input file.
Example: {'input_path': 'folder/input/notebook.ipynb', 'input_params': {'param1': 'value1'}}
:param output_config: Configuration for the output format. It should include an output_formats parameter to specify the output format.
Example: {'output_formats': ['NOTEBOOK']}
:param compute: compute configuration to use for the notebook execution. This is a required attribute if the execution is on a remote compute.
Example: {"instance_type": "ml.m5.large", "volume_size_in_gb": 30, "volume_kms_key_id": "", "image_details": {"ecr_uri": "string"}, "container_entrypoint": ["string"]}
:param domain_region: The AWS region for the domain. If not provided, the default AWS region will be used.
:param compute: compute configuration to use for the notebook execution. This is a required attribute
if the execution is on a remote compute.
Example::

{
"instance_type": "ml.c5.xlarge",
"image_details": {
"image_name": "sagemaker-distribution-prod",
"image_version": "3",
"ecr_uri": "123456123456.dkr.ecr.us-west-2.amazonaws.com/ImageName:latest",
},
}

:param termination_condition: conditions to match to terminate the remote execution.
Example: {"MaxRuntimeInSeconds": 3600}
Example: ``{"MaxRuntimeInSeconds": 3600}``
:param tags: tags to be associated with the remote execution runs.
Example: {"md_analytics": "logs"}
Example: ``{"md_analytics": "logs"}``
:param waiter_delay: Interval in seconds to check the task execution status.
:param waiter_max_attempts: Number of attempts to wait before returning FAILED.
"""
Expand All @@ -66,7 +85,10 @@ def __init__(
self,
execution_name: str,
input_config: dict | None = None,
domain_id: str | None = None,
project_id: str | None = None,
output_config: dict | None = None,
domain_region: str | None = None,
compute: dict | None = None,
termination_condition: dict | None = None,
tags: dict | None = None,
Expand All @@ -78,6 +100,9 @@ def __init__(
super().__init__(*args, **kwargs)
self._sagemaker_studio = SageMakerStudioAPI(self._get_sagemaker_studio_config())
self.execution_name = execution_name
self.domain_id = domain_id
self.project_id = project_id
self.domain_region = domain_region
self.input_config = input_config or {}
self.output_config = output_config or {"output_formats": ["NOTEBOOK"]}
self.compute = compute
Expand Down Expand Up @@ -114,17 +139,20 @@ def start_notebook_execution(self):
start_execution_params = {
"execution_name": self.execution_name,
"execution_type": "NOTEBOOK",
"domain_id": self.domain_id,
"project_id": self.project_id,
"input_config": self._format_start_execution_input_config(),
"output_config": self._format_start_execution_output_config(),
"termination_condition": self.termination_condition,
"tags": self.tags,
}

if self.domain_region:
start_execution_params["domain_region"] = self.domain_region

if self.compute:
start_execution_params["compute"] = self.compute
else:
start_execution_params["compute"] = {"instance_type": "ml.m6i.xlarge"}

print(start_execution_params)
return self._sagemaker_studio.execution_client.start_execution(**start_execution_params)

def wait_for_execution_completion(self, execution_id, context):
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -48,8 +48,11 @@ class SageMakerNotebookOperator(BaseOperator):

notebook_operator = SageMakerNotebookOperator(
task_id="notebook_task",
domain_id="dzd-example123456",
project_id="example123456",
input_config={"input_path": "path/to/notebook.ipynb", "input_params": ""},
output_config={"output_format": "ipynb"},
domain_region="us-east-1",
wait_for_completion=True,
waiter_delay=10,
waiter_max_attempts=1440,
Expand All @@ -63,12 +66,28 @@ class SageMakerNotebookOperator(BaseOperator):
:param output_config: Configuration for the output format. It should include an output_format parameter to control
the format of the notebook execution output.
Example: {"output_formats": ["NOTEBOOK"]}
:param compute: compute configuration to use for the notebook execution. This is a required attribute if the execution is on a remote compute.
Example: {"instance_type": "ml.m5.large", "volume_size_in_gb": 30, "volume_kms_key_id": "", "image_details": {"ecr_uri": "string"}, "container_entrypoint": ["string"]}
:param domain_id: The domain ID for Amazon SageMaker Unified Studio. Optional - if not provided,
the SDK will attempt to resolve it from the MWAA environment.
:param project_id: The project ID for Amazon SageMaker Unified Studio. Optional - if not provided,
the SDK will attempt to resolve it from the MWAA environment.
:param domain_region: The AWS region for the domain. If not provided, the default AWS region will be used.
:param compute: compute configuration to use for the artifact execution. This is a required attribute
if the execution is on a remote compute.
Example::

{
"instance_type": "ml.c5.xlarge",
"image_details": {
"image_name": "sagemaker-distribution-prod",
"image_version": "3",
"ecr_uri": "123456123456.dkr.ecr.us-west-2.amazonaws.com/ImageName:latest"
},
}

:param termination_condition: conditions to match to terminate the remote execution.
Example: { "MaxRuntimeInSeconds": 3600 }
Example: ``{"MaxRuntimeInSeconds": 3600}``
:param tags: tags to be associated with the remote execution runs.
Example: { "md_analytics": "logs" }
Example: ``{"md_analytics": "logs"}``
:param wait_for_completion: Indicates whether to wait for the notebook execution to complete. If True, wait for completion; if False, don't wait.
:param waiter_delay: Interval in seconds to check the notebook execution status.
:param waiter_max_attempts: Number of attempts to wait before returning FAILED.
Expand All @@ -87,7 +106,10 @@ def __init__(
self,
task_id: str,
input_config: dict,
domain_id: str | None = None,
project_id: str | None = None,
output_config: dict | None = None,
domain_region: str | None = None,
compute: dict | None = None,
termination_condition: dict | None = None,
tags: dict | None = None,
Expand All @@ -99,8 +121,11 @@ def __init__(
):
super().__init__(task_id=task_id, **kwargs)
self.execution_name = task_id
self.domain_id = domain_id
self.project_id = project_id
self.input_config = input_config
self.output_config = output_config or {"output_formats": ["NOTEBOOK"]}
self.domain_region = domain_region
self.compute = compute or {}
self.termination_condition = termination_condition or {}
self.tags = tags or {}
Expand All @@ -119,9 +144,12 @@ def notebook_execution_hook(self):
raise AirflowException("input_path is a required field in the input_config")

return SageMakerNotebookHook(
domain_id=self.domain_id,
project_id=self.project_id,
input_config=self.input_config,
output_config=self.output_config,
execution_name=self.execution_name,
domain_region=self.domain_region,
compute=self.compute,
termination_condition=self.termination_condition,
tags=self.tags,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -200,28 +200,6 @@ def test_set_xcom_s3_path_negative_missing_context(self):
with pytest.raises(AirflowException, match="context is required"):
self.hook._set_xcom_s3_path(self.s3Path, {})

def test_start_notebook_execution_default_compute(self):
Comment thread
vincbeck marked this conversation as resolved.
"""Test that default compute uses ml.m6i.xlarge instance type."""
hook_without_compute = SageMakerNotebookHook(
input_config={
"input_path": "test-data/notebook/test_notebook.ipynb",
"input_params": {"key": "value"},
},
output_config={"output_formats": ["NOTEBOOK"]},
execution_name="test-execution",
waiter_delay=10,
)
hook_without_compute._sagemaker_studio = MagicMock()
hook_without_compute._sagemaker_studio.execution_client = MagicMock(spec=ExecutionClient)
hook_without_compute._sagemaker_studio.execution_client.start_execution.return_value = {
"executionId": "123456"
}

hook_without_compute.start_notebook_execution()

call_kwargs = hook_without_compute._sagemaker_studio.execution_client.start_execution.call_args[1]
assert call_kwargs["compute"] == {"instance_type": "ml.m6i.xlarge"}

def test_start_notebook_execution_custom_compute(self):
"""Test that custom compute config is used when provided."""
custom_compute = {"instance_type": "ml.c5.xlarge", "volume_size_in_gb": 50}
Expand Down
Loading