使用 AWS SageMaker、AWS StepFunctions、AWS Eventbridge 自动化机器学习工作流程答案

【问题标题】：Automating a Machine Learning workflow with AWS SageMaker, AWS StepFunctions, AWS Eventbridge使用 AWS SageMaker、AWS StepFunctions、AWS Eventbridge 自动化机器学习工作流程
【发布时间】：2022-10-06 12:36:37
【问题描述】：

我正在 AWS SageMaker 中创建一个简单的 ML 工作流，并尝试使用 AWS StepFunctions 对其进行编排。目前，我能够使用 Python SDK 创建一个步骤并在 SageMaker Notebook 中执行。

最终，我想使用 AWS eventbridge 每周自动调用一次。当我使用从 sagemaker 成功运行后创建的相同状态机在 eventbridge 中测试一个简单的规则时，stepfunctions 失败。当前的 Python SDK 没有告诉我们如何自动化。

任何人都可以建议/告诉我如何使这种自动化工作。为简单起见，我在下面包含了一个单步工作流程：感谢大家！

# Install necessary upgrades
import sys
!{sys.executable} -m pip install --upgrade pip
!{sys.executable} -m pip install --upgrade stepfunctions

# import necessary libraries .. 
# e.g ...
import boto3
import sagemaker
import stepfunctions

# There are many more libraries not shown here...

接下来我定义了一个简单的处理步骤如下：

execution_input = ExecutionInput(
    schema={
        \"PreprocessingJobName\": str,
    }
)

# Using Script processing

inputs=[
    ProcessingInput(
        source=input_data_path, 
        destination=\"/opt/ml/processing/input\",
        input_name=\"input_data\"
    ),
    ProcessingInput(
        source=\'s3://{}/{}\'.format(bucket, \'script_processing/code/preprocessing.py\'), 
        destination=\"/opt/ml/processing/input/code\",
        input_name=\"code\",
    ),
]

outputs=[
    ProcessingOutput(output_name=\"train_data\",
                     source=\"/opt/ml/processing/output/train\",
                     destination=\"{}/{}\".format(output_data_path, \"train_data\")),
    ProcessingOutput(output_name=\"test_data\",
                     source=\"/opt/ml/processing/output/test\",
                     destination=\"{}/{}\".format(output_data_path, \"test_data\")),
]

# Create a SageMaker ScriptProcessor instance
script_processor = ScriptProcessor(
    command=[\'python3\'],
    image_uri=preprocessing_image,
    role=sagemaker_execution_role,
    instance_count=1,
    instance_type=\"ml.m5.xlarge\",
)

# Create Processing Step
processing_step = ProcessingStep(
    \"my-processing-step\",
    processor=script_processor,
    job_name=execution_input[\"PreprocessingJobName\"],    
    inputs=inputs,
    outputs=outputs,
    container_arguments=[\"--train-test-split-ratio\", \"0.2\"],
    container_entrypoint=[\"python3\", \"/opt/ml/processing/input/code/preprocessing.py\"],
)

最后整理一下步骤：

workflow_graph = Chain(
    [processing_step]
    )

# Next, we define the workflow
branching_workflow = Workflow(
    name = \"MyWorkflow-processing\", 
    definition = workflow_graph,
    role = workflow_execution_role
)

branching_workflow.create()


# Execute the workflow
workflow_execution = branching_workflow.execute(
    inputs = { 
        \"PreprocessingJobName\": \"preprocess-{}-{}\".format(uuid.uuid1().hex),        
    }
)
execution_output = workflow_execution.get_output(wait=True)

从 Jupyter Notebook 执行后，它运行成功。

您想每周安排一次 aws step function 吗？
当 step 函数失败时，您会看到什么错误？
@SUDARSHAN，是的，我想安排这个工作流程每周触发一次。
@KiritThadaka，我收到如下错误： { \"error\": \"States.Runtime\", \"cause\": \"执行状态 \'my-processing-step\' 时发生错误（输入于事件 ID #2)。在输入 \'{\\\' 中找不到为字段 \'ProcessingJobName.$\' 指定的 JSONPath \'$$.Execution.Input[\'PreprocessingJobName\']\' "执行\\\":{\\\"Id\\\":\\\"arn:aws:states:us-west-###########\\:执行:MyWorkflow-加工.......
如果你在我的帖子中查看上面的 Jupyter sn-p，有一行代码说：``` execution_input = ExecutionInput( schema={ \"PreprocessingJobName\": str, } ) ``` 我的猜测是什么时候状态机在 Jupyter 之外触发，此参数未传递到处理步骤。如何确保这是参数被传递到流程中？

标签： amazon-web-services aws-lambda amazon-sagemaker aws-step-functions aws-event-bridge

【解决方案1】：

如果调度是您面临的问题，那么您可以使用这个 cron

创建 EventBridge 规则

cron 表达式

0 1 ? * 周六 *
1. 从 aws lambda 函数启动 step 函数的示例代码
```
client = boto3.client('stepfunctions')
 workflowInput={
 'ID' : ID
 }

 step_function='Give unique name to setp function exceution'
 response = client.start_execution(
 stateMachineArn='arn:aws:states:***************',
 name=step_function,
 input= json.dumps(workflowInput)
 )
```
workflowInput 输入将是您的 step 函数的输入。如果您仔细输入，您所面临的错误也可以得到解决。它接受 JSON 格式的输入。您可以继续向 json 添加更多元素。

在 step 函数中管理长时间运行的任务。

Lambda 将调用 step 函数并且不会以任何方式等待其完成。您可以在您的用例中做的是，您可以使用另一个事件桥来更改步骤功能状态（SUCCEEDED、FAILED、TIMED_OUT 或 ABORTED）。一旦您的步进函数成功，lambda 将调用并通知所有详细信息或事件，您也可以拥有步进函数的执行历史记录。

我希望这对你有用。

【讨论】：

您好，感谢您的提示。但我担心，这可能不适用于我的情况，因为 lambda 限制 < 15 分钟。我的整个工作流程可以超过一小时。请问您是否知道我们如何使用 Lambda 函数在 sagemaker 实例中运行笔记本？场景：我有一个 Sagemaker 实例，它有多个书籍。但是是否可以旋转该实例，然后从 lambda 运行“特定”笔记本？非常感谢！