翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。 # @step デコレータを使用した Python コードのリフトアンドシフト `@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 decorator sample notebooks](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) + [ ## `@step` でデコレートしたステップで `ConditionStep` でを使用する ](#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` インスタンスは、その関数の実際の戻り値のプロキシです。`DelayedReturn` インスタンスは、別の関数に引数として渡すことも、ステップとしてパイプラインインスタンスに直接渡すこともできます。`DelayedReturn` クラスの詳細については、「[sagemaker.workflow.function\$1step.DelayedReturn](https://sagemaker.readthedocs.io/en/stable/workflows/pipelines/sagemaker.workflow.pipelines.html#sagemaker.workflow.function_step.DelayedReturn)」を参照してください。 ## ステップ間の依存関係を作成する 2 つのステップ間に依存関係を作成する際は、パイプライングラフのステップ間に接続を作成します。以下のセクションでは、パイプラインステップ間の依存関係を作成する複数の方法を紹介します。 ### 入力引数を使用したデータ依存関係 ある関数の `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, ) ``` ## `@step` でデコレートしたステップで `ConditionStep` でを使用する Pipelines は、先行するステップの結果を評価してパイプラインで実行するアクションを決定する、`ConditionStep` クラスをサポートしています。`ConditionStep` は、`@step` でデコレートしたステップでも使用できます。`ConditionStep` で任意の `@step` でデコレートしたステップの出力を使用するには、そのステップの出力を `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 リソースで、またはローカルでパイプラインを実行する方法について説明します。 従来の SageMaker AI パイプライン実行と同様に、`pipeline.start()` 関数を使用して新しいパイプライン実行を開始します。`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)」を参照してください。ローカルモードを使用するには、次の例に示されるとおり、パイプライン定義に `SageMakerSession` ではなく、`LocalPipelineSession` を指定します。 ``` 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" ``` `CustomFileFilter` で `IncludeLocalWorkdir` を使用する方法の詳細については、「[モジュラーコードと @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` デコレータを使用する際に従うべきベストプラクティスを提案しています。 ## ウォームプールを使用する パイプライン ステップの実行を高速化するには、トレーニングジョブ用に提供されているウォームプール機能を使用します。次のスニペットに示されるとおり、`@step` デコレータに `keep_alive_period_in_seconds` 引数を指定することで、ウォームプール機能を有効にできます。 ``` @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)」の「*Data serialization and deserialization*」セクションを参照してください。`@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` デコレータは、以下のパイプライン機能では使用できません。 + [Pipeline ステップのキャッシュ](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)