本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。 # 使用 @step 裝飾項目直接移轉程式碼 `@step` 裝飾項目是將本機機器學習 (ML) 程式碼轉換為一或多個管道步驟的特徵。您可以像撰寫任何 ML 專案一樣撰寫 ML 函式。一旦使用 `@remote` 裝飾項目在本機或做為訓練任務測試，您就可以透過新增 `@step` 裝飾項目，將函式轉換為 SageMaker AI 管道步驟。然後，您可以將 `@step` 裝飾的函式呼叫的輸出做為步驟傳遞至 Pipelines，以建立和執行管道。您也可以使用 `@step` 裝飾項目鏈結一系列函式，以建立多步驟有向無環圖 (DAG) 管道。 使用 `@step` 裝飾項目的設定與使用 `@remote` 裝飾項目的設定相同。如需如何[設定環境](https://docs.aws.amazon.com/sagemaker/latest/dg/train-remote-decorator.html#train-remote-decorator-env)和[使用組態檔案](https://docs.aws.amazon.com/sagemaker/latest/dg/train-remote-decorator-config.html)來設定預設值的詳細資訊，請參閱遠端函式文件。如需 `@step` 裝飾項目的詳細資訊，請參閱。[sagemaker.workflow.function\$1step.step](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.function_step.step)。 若要檢視示範如何使用 `@step` 裝飾項目的範例筆記本，請參閱 [@step 裝飾項目範例筆記本](https://github.com/aws/amazon-sagemaker-examples/tree/main/sagemaker-pipelines/step-decorator)。 下列各節說明如何使用 `@step` 裝飾項目註釋您的本機 ML 程式碼，以建立步驟、使用步驟建立和執行管道，以及為您的使用案例自訂體驗。 **Topics** + [使用 `@step` 裝飾的函式建立管道](pipelines-step-decorator-create-pipeline.md) + [執行管道](pipelines-step-decorator-run-pipeline.md) + [設定您的管道](pipelines-step-decorator-cfg-pipeline.md) + [最佳實務](pipelines-step-decorator-best.md) + [限制](pipelines-step-decorator-limit.md) # 使用 `@step` 裝飾的函式建立管道 您可以建立一個管道，方法是使用 `@step` 裝飾項目將 Python 函式轉換為管道步驟、在這些函式之間建立相依性以建立管道圖形 (或有向無環圖 (DAG))，然後將該圖形的分葉節點做為步驟清單傳遞至管道。下列各節會使用範例詳細說明此程序。 **Topics** + [將函式轉換為步驟](#pipelines-step-decorator-run-pipeline-convert) + [在步驟之間建立相依性](#pipelines-step-decorator-run-pipeline-link) + [使用 `ConditionStep` 搭配 `@step` 裝飾的步驟](#pipelines-step-decorator-condition) + [使用步驟的 `DelayedReturn` 輸出定義管道](#pipelines-step-define-delayed) + [建立管道](#pipelines-step-decorator-pipeline-create) ## 將函式轉換為步驟 若要使用 `@step` 裝飾項目建立步驟，請使用 `@step` 註釋函式。下列範例顯示預先處理資料的 `@step` 裝飾函式。 ``` from sagemaker.workflow.function_step import step @step def preprocess(raw_data): df = pandas.read_csv(raw_data) ... return procesed_dataframe step_process_result = preprocess(raw_data) ``` 當您調用 `@step` 裝飾函式時，SageMaker AI 會傳回 `DelayedReturn` 執行個體，而不是執行該函式。`DelayedReturn` 執行個體是該函式實際傳回的 Proxy。`DelayedReturn` 執行個體可以做為引數傳遞至另一個函式，或直接做為步驟傳遞至管道執行個體。如需 `DelayedReturn` 類別的相關資訊，請參閱 [sagemaker.workflow.function\$1step.DelayedReturn](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.function_step.DelayedReturn)。 ## 在步驟之間建立相依性 當您在兩個步驟之間建立相依性時，您可以在管道圖中的步驟之間建立連線。下列各節介紹在管道步驟之間建立相依性的多種方式。 ### 透過輸入引數的資料相依性 將某個函式的 `DelayedReturn` 輸出做為輸入傳遞至另一個函式，會自動在管道 DAG 中建立資料相依性。在下列範例中，將 `preprocess` 函式的 `DelayedReturn` 輸出傳遞至 `train` 函式，會在 `preprocess` 與 `train` 之間建立相依性。 ``` from sagemaker.workflow.function_step import step @step def preprocess(raw_data): df = pandas.read_csv(raw_data) ... return procesed_dataframe @step def train(training_data): ... return trained_model step_process_result = preprocess(raw_data) step_train_result = train(step_process_result) ``` 前一個範例定義了使用 `@step` 裝飾的訓練函式。調用此函式時，其會接收預先處理管道步驟的 `DelayedReturn` 輸出做為輸入。調用訓練函式會傳回另一個 `DelayedReturn`。執行個體。此執行個體會保留形成管道 DAG 之函式 (即此範例中 `preprocess` 的步驟) 中定義之所有先前步驟的相關資訊。 在前一個範例中，`preprocess` 函式會傳回單一值。如需清單或元組等更複雜的傳回類型，請參閱[限制](pipelines-step-decorator-limit.md)。 ### 定義自訂相依性 在前一個範例中，`train` 函式收到 `preprocess` 的 `DelayedReturn` 輸出並建立了相依性。如果您想要明確定義相依性，而不傳遞前一個步驟輸出，請使用 `add_depends_on` 函式搭配步驟。您可以使用 `get_step()` 函式，從其 `DelayedReturn` 執行個體擷取基礎步驟，然後使用相依性做為輸入來呼叫 `add_depends_on`\$1on。若要檢視 `get_step()` 函式定義，請參閱 [sagemaker.workflow.step\$1outputs.get\$1step](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.step_outputs.get_step)。下列範例展示如何使用 `get_step()` 和 `add_depends_on()` 建立 `preprocess` 與 `train` 之間的相依性。 ``` from sagemaker.workflow.step_outputs import get_step @step def preprocess(raw_data): df = pandas.read_csv(raw_data) ... processed_data = .. return s3.upload(processed_data) @step def train(): training_data = s3.download(....) ... return trained_model step_process_result = preprocess(raw_data) step_train_result = train() get_step(step_train_result).add_depends_on([step_process_result]) ``` ### 將資料從 `@step` 裝飾函式傳遞至傳統管道步驟，或從中將資料傳遞至該函式 您可以建立一個管道，其中包含 `@step` 裝飾步驟和傳統管道步驟，以及在它們之間傳遞資料。例如，您可以使用 `ProcessingStep` 來處理資料，並將其結果傳遞至 `@step` 裝飾的訓練函式。在下列範例中，`@step` 裝飾的訓練步驟會參考處理步驟的輸出。 ``` # Define processing step from sagemaker.sklearn.processing import SKLearnProcessor from sagemaker.processing import ProcessingInput, ProcessingOutput from sagemaker.workflow.steps import ProcessingStep sklearn_processor = SKLearnProcessor( framework_version='1.2-1', role='arn:aws:iam::123456789012:role/SagemakerExecutionRole', instance_type='ml.m5.large', instance_count='1', ) inputs = [ ProcessingInput(source=input_data, destination="/opt/ml/processing/input"), ] outputs = [ ProcessingOutput(output_name="train", source="/opt/ml/processing/train"), ProcessingOutput(output_name="validation", source="/opt/ml/processing/validation"), ProcessingOutput(output_name="test", source="/opt/ml/processing/test") ] process_step = ProcessingStep( name="MyProcessStep", step_args=sklearn_processor.run(inputs=inputs, outputs=outputs,code='preprocessing.py'), ) ``` ``` # Define a @step-decorated train step which references the # output of a processing step @step def train(train_data_path, test_data_path): ... return trained_model step_train_result = train( process_step.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri, process_step.properties.ProcessingOutputConfig.Outputs["test"].S3Output.S3Uri, ) ``` ## 使用 `ConditionStep` 搭配 `@step` 裝飾的步驟 Pipelines 支援 `ConditionStep` 類別，其會評估先前步驟的結果，以決定要在管道中採取的動作。您也可以使用 `ConditionStep` 搭配 `@step` 裝飾的步驟。若要使用任何 `@step` 裝飾步驟的輸出搭配 `ConditionStep`，請輸入該步驟的輸出做為 `ConditionStep` 的引數。在下列範例中，條件步驟會收到 `@step` 裝飾的模型評估步驟的輸出。 ``` # Define steps @step(name="evaluate") def evaluate_model(): # code to evaluate the model return { "rmse":rmse_value } @step(name="register") def register_model(): # code to register the model ... ``` ``` # Define ConditionStep from sagemaker.workflow.condition_step import ConditionStep from sagemaker.workflow.conditions import ConditionGreaterThanOrEqualTo from sagemaker.workflow.fail_step import FailStep conditionally_register = ConditionStep( name="conditional_register", conditions=[ ConditionGreaterThanOrEqualTo( # Output of the evaluate step must be json serializable left=evaluate_model()["rmse"], # right=5, ) ], if_steps=[FailStep(name="Fail", error_message="Model performance is not good enough")], else_steps=[register_model()], ) ``` ## 使用步驟的 `DelayedReturn` 輸出定義管道 無論您是否使用 `@step` 裝飾項目，您都會以相同的方式定義管道。將 `DelayedReturn` 執行個體傳遞至您的管道時，您不需要傳遞完整步驟清單以建置管道。SDK 會根據您定義的相依性，自動推斷先前的步驟。管道圖中包含您傳遞至管道的 `Step` 物件或 `DelayedReturn` 物件的所有先前步驟。在下列範例中，管道會收到 `train` 函式的 `DelayedReturn` 物件。SageMaker AI 會將 `preprocess` 步驟做為 `train` 的前一個步驟傳遞至管道圖。 ``` from sagemaker.workflow.pipeline import Pipeline pipeline = Pipeline( name="", steps=[step_train_result], sagemaker_session=, ) ``` 如果步驟之間沒有資料或自訂相依性，而且您平行執行多個步驟，則管道圖具有多個分葉節點。將清單中的所有這些分葉節點傳遞至管道定義中的 `steps` 引數，如下列範例所示： ``` @step def process1(): ... return data @step def process2(): ... return data step_process1_result = process1() step_process2_result = process2() pipeline = Pipeline( name="", steps=[step_process1_result, step_process2_result], sagemaker_session=sagemaker-session, ) ``` 當管道執行時，這兩個步驟都會平行執行。 您只能將圖形的分葉節點傳遞至管道，因為分葉節點包含透過資料或自訂相依性定義之所有先前步驟的相關資訊。當其編譯管道時，SageMaker AI 也會推斷形成管道圖的所有後續步驟，並將每個步驟做為個別步驟新增至管道。 ## 建立管道 呼叫 `pipeline.create()` 來建立管道，如下列程式碼片段所示。如需 `create()` 的詳細資訊，請參閱 [sagemaker.workflow.pipeline.Pipeline.create](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.pipeline.Pipeline.create)。 ``` role = "pipeline-role" pipeline.create(role) ``` 當您呼叫 `pipeline.create()` 時，SageMaker AI 會編譯所有定義為管道執行個體一部分的步驟。SageMaker AI 會將序列化函式、引數和所有其他步驟相關成品上傳至 Amazon S3。 根據下列結構，資料位於 S3 儲存貯體中： ``` s3_root_uri/ pipeline_name/ sm_rf_user_ws/ workspace.zip # archive of the current working directory (workdir) step_name/ timestamp/ arguments/ # serialized function arguments function/ # serialized function pre_train_dependencies/ # any dependencies and pre_execution scripts provided for the step execution_id/ step_name/ results # returned output from the serialized function including the model ``` `s3_root_uri` 定義在 SageMaker AI 組態檔案中，並套用至整個管道。如果未定義，則會使用預設的 SageMaker AI 儲存貯體。 **注意** 每次 SageMaker AI 編譯管道時，SageMaker AI 都會將步驟的序列化函式、引數和相依性儲存在以目前時間加上時間戳記的資料夾中。每次執行 `pipeline.create()`、`pipeline.update()`、`pipeline.upsert()` 或 `pipeline.definition()` 時都會發生這種情況。 # 執行管道 下頁描述如何使用 Amazon SageMaker Pipelines 執行管道，無論是搭配 SageMaker AI 資源還是在本機。 使用 `pipeline.start()` 函式啟動新的管道執行，就像傳統 SageMaker AI 管道執行一樣。如需 `start()` 函式的相關資訊，請參閱 [sagemaker.workflow.pipeline.Pipeline.start](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.pipeline.Pipeline.start)。 **注意** 使用 `@step` 裝飾項目定義的步驟會做為訓練任務執行。因此，請注意下列限制： 您帳戶中的執行個體限制和訓練任務限制。相應地更新您的限制，以避免發生任何限流或資源限制問題。 與管道中每個訓練步驟執行相關聯的金錢成本。如需詳細資訊，請參閱 [Amazon SageMaker 定價](https://aws.amazon.com/sagemaker/pricing/)。 ## 從本機執行的管道擷取結果 若要檢視管道執行任何步驟的結果，請使用 [execution.result()](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.pipeline._PipelineExecution.result )，如下列程式碼片段所示： ``` execution = pipeline.start() execution.result(step_name="train") ``` **注意** Pipelines 不支援本機模式下的 `execution.result()`。 您一次只能擷取一個步驟的結果。如果步驟名稱是由 SageMaker AI 產生，您可以呼叫 `list_steps` 來擷取步驟名稱，如下所示： ``` execution.list_step() ``` ## 在本機執行管道 您可以像執行傳統管道步驟一樣，使用 `@step` 裝飾步驟在本機執行管道。如需本機模式管道執行的詳細資訊，請參閱[使用本機模式執行管道](pipelines-local-mode.md)。若要使用本機模式，請將 `LocalPipelineSession` 而非 `SageMakerSession` 提供給管道定義，如下列範例所示： ``` from sagemaker.workflow.function_step import step from sagemaker.workflow.pipeline import Pipeline from sagemaker.workflow.pipeline_context import LocalPipelineSession @step def train(): training_data = s3.download(....) ... return trained_model step_train_result = train() local_pipeline_session = LocalPipelineSession() local_pipeline = Pipeline( name="", steps=[step_train_result], sagemaker_session=local_pipeline_session # needed for local mode ) local_pipeline.create(role_arn="role_arn") # pipeline runs locally execution = local_pipeline.start() ``` # 設定您的管道 建議您使用 SageMaker AI 組態檔案來設定管道的預設值。如需 SageMaker AI 組態檔案的詳細資訊，請參閱[搭配 SageMaker Python SDK 設定和使用預設值](https://sagemaker.readthedocs.io/en/stable/overview.html#configuring-and-using-defaults-with-the-sagemaker-python-sdk)。新增至組態檔案的任何組態都會套用至管道中的所有步驟。如果您想要覆寫任何步驟的選項，請在 `@step` 裝飾項目引數中提供新值。下列主題描述如何設定組態檔案。 組態檔案中 `@step` 裝飾項目的組態與 `@remote` 裝飾項目的組態相同。若要在組態檔案中設定管道角色 ARN 和管道標籤，請使用下列程式碼片段中顯示的 `Pipeline` 區段： ``` SchemaVersion: '1.0' SageMaker: Pipeline: RoleArn: 'arn:aws:iam::555555555555:role/IMRole' Tags: - Key: 'tag_key' Value: 'tag_value' ``` 對於您在組態檔案中設定的大多數預設值，您也可以透過將新值傳遞至 `@step` 裝飾項目來覆寫這些預設值。例如，您可以覆寫組態檔案中為預先處理步驟設定的執行個體類型，如下列範例所示： ``` @step(instance_type="ml.m5.large") def preprocess(raw_data): df = pandas.read_csv(raw_data) ... return procesed_dataframe ``` 一些引數不是 `@step` 裝飾項目參數清單的一部分 - 只能透過 SageMaker AI 組態檔案為整個管道設定這些引數。它們列示如下： + `sagemaker_session` (`sagemaker.session.Session`)：獲 SageMaker AI 委派服務呼叫的基礎 SageMaker AI 工作階段。如果未指定，則會使用預設組態建立工作階段，如下所示： ``` SageMaker: PythonSDK: Modules: Session: DefaultS3Bucket: 'default_s3_bucket' DefaultS3ObjectKeyPrefix: 'key_prefix' ``` + `custom_file_filter` (`CustomFileFilter)`)：一種 `CustomFileFilter` 物件，其指定要包含在管道步驟中的本機目錄和檔案。如果未指定，此值預設為 `None`。`custom_file_filter` 若要生效，您必須將 `IncludeLocalWorkdir` 設定為 `True`。下列範例顯示的組態會忽略所有筆記本檔案，以及名為 `data` 的檔案和目錄。 ``` SchemaVersion: '1.0' SageMaker: PythonSDK: Modules: RemoteFunction: IncludeLocalWorkDir: true CustomFileFilter: IgnoreNamePatterns: # files or directories to ignore - "*.ipynb" # all notebook files - "data" # folder or file named "data" ``` 如需如何使用 `IncludeLocalWorkdir` 搭配 `CustomFileFilter` 的詳細資訊，請參閱[搭配 @remote 裝飾項目使用模組化代碼](train-remote-decorator-modular.md)。 + `s3_root_uri (str)`：SageMaker AI 將程式碼封存和資料上傳至其中的根 Amazon S3 資料夾。如果未指定，則會使用預設 SageMaker AI 儲存貯體。 + `s3_kms_key (str)`：用來加密輸入和輸出資料的金鑰。您只能在 SageMaker AI 組態檔案中設定此引數，而且此引數會套用至管道中定義的所有步驟。如果未指定，則值預設為 `None`。如需範例 S3 KMS 金鑰組態，請參閱下列程式碼片段： ``` SchemaVersion: '1.0' SageMaker: PythonSDK: Modules: RemoteFunction: S3KmsKeyId: 's3kmskeyid' S3RootUri: 's3://amzn-s3-demo-bucket/my-project ``` # 最佳實務 以下各節建議在您針對管道步驟使用 `@step` 裝飾項目時要遵循的最佳實務。 ## 使用暖集區 如需加快管道步驟的執行速度，請使用為訓練任務提供的暖集區功能。您可以將 `keep_alive_period_in_seconds` 引數提供給 `@step` 裝飾項目以開啟暖集區功能，如下列程式碼片段所示範： ``` @step( keep_alive_period_in_seconds=900 ) ``` 如需有關暖集區的詳細資訊，請參閱 [SageMaker AI 受管暖集區](train-warm-pools.md)。 ## 建構您的目錄 建議您在使用 `@step` 裝飾項目時使用程式碼模組。將您在其中調用步驟函式並定義管道的 `pipeline.py` 模組放在工作區的根目錄。建議的結構如下所示： ``` . ├── config.yaml # the configuration file that define the infra settings ├── requirements.txt # dependencies ├── pipeline.py # invoke @step-decorated functions and define the pipeline here ├── steps/ | ├── processing.py | ├── train.py ├── data/ ├── test/ ``` # 限制 下列各節概述您在針對管道步驟中使用 `@step` 裝飾項目時應注意的限制。 ## 函式引數限制 將輸入引數傳遞至 `@step` 裝飾函式時，適用下列限制： + 您可以將 `DelayedReturn`、`Properties` (其他類型的步驟)、`Parameter` 和 `ExecutionVariable` 物件傳遞至 `@step` 裝飾函式做為引數。但是 `@step` 裝飾函式不支援 `JsonGet` 和 `Join` 物件做為引數。 + 您無法直接從 `@step` 函式存取管道變數。以下範例產生錯誤： ``` param = ParameterInteger(name="", default_value=10) @step def func(): print(param) func() # this raises a SerializationError ``` + 您無法將一個管道變數套疊另一個物件中，然後將其傳遞至 `@step` 函式。以下範例產生錯誤： ``` param = ParameterInteger(name="", default_value=10) @step def func(arg): print(arg) func(arg=(param,)) # this raises a SerializationError because param is nested in a tuple ``` + 由於函式的輸入和輸出是序列化的，因此對可做為函數輸入或輸出傳遞的資料類型有限制。如需詳細資訊，請參閱[調用遠端函式](train-remote-decorator-invocation.md)的*資料序列化和還原序列化*一節。相同的限制適用於 `@step` 裝飾函式。 + 任何具有 boto 用戶端的物件都無法序列化，因此您無法將這類物件做為輸入傳遞至 `@step` 裝飾函式，也無法將其做為此函式的輸出傳遞。例如，`Estimator`、`Predictor` 和 `Processor` 等 SageMaker Python SDK 用戶端類別無法序列化。 ## 函式匯入 您應該在函式內匯入步驟所需的程式庫，而不是函數外。如果您在全域範圍內匯入它們，則在序列化函式時您會有發生匯入衝突的風險。例如，`sklearn.pipeline.Pipeline` 可被 `sagemaker.workflow.pipeline.Pipeline` 覆寫。 ## 參考函式傳回值的子成員 如果您參考 `@step` 裝飾函式傳回值的子成員，則適用下列限制： + 如果 `DelayedReturn` 物件代表元組、清單或字典，您可以參考具有 `[]` 的子成員，如下列範例所示： ``` delayed_return[0] delayed_return["a_key"] delayed_return[1]["a_key"] ``` + 您無法解壓縮元組或清單輸出，因為當您調用函式時，無法得知基礎元組或清單的確切長度。以下範例產生錯誤： ``` a, b, c = func() # this raises ValueError ``` + 您無法逐一查看 `DelayedReturn` 物件。以下範例會引發錯誤： ``` for item in func(): # this raises a NotImplementedError ``` + 您無法參考具有 '`.`' 的任意子成員。以下範例產生錯誤： ``` delayed_return.a_child # raises AttributeError ``` ## 不支援的現有管道特徵 您無法使用 `@step` 裝飾項目搭配下列管道特徵： + [管道步驟快取](https://docs.aws.amazon.com/sagemaker/latest/dg/pipelines-caching.html) + [屬性檔案](https://docs.aws.amazon.com/sagemaker/latest/dg/build-and-manage-propertyfile.html#build-and-manage-propertyfile-property)