特征处理器管道的计划执行和基于事件的执行

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

可以将 Amazon F SageMaker eature Store 功能处理管道执行配置为根据预先配置的计划自动和异步启动，也可以根据其他 AWS 服务事件的结果自动和异步启动。例如，您可以计划在每个月的第一天执行特征处理管道，或者将两个管道链接在一起，以便在源管道执行完成后自动执行目标管道。

基于计划的执行

功能处理器 SDK 提供了一schedule个 API，用于通过 Amazon S EventBridge cheduler 集成，定期运行功能处理器管道。可以使用ScheduleExpression参数使用、或cron表达式来指定计划 atrate，其表达式与 Amazon 支持的表达式相同 EventBridge。从语义上讲，计划 API 是一项更新插入操作，即如果已经存在计划，它会更新计划；否则，它会创建计划。有关 EventBridge 表达式和示例的更多信息，请参阅《日程 EventBridge 安排 EventBridge 器用户指南》中的日程安排类型。

以下示例使用特征处理器 schedule API，该 API 使用 at、rate 和 cron 表达式。

from sagemaker.feature_store.feature_processor import schedule
pipeline_name='feature-processor-pipeline'

event_bridge_schedule_arn = schedule(
    pipeline_name=pipeline_name, 
    schedule_expression="at(2020-11-30T00:00:00)"
)

event_bridge_schedule_arn = schedule(
    pipeline_name=pipeline_name, 
    schedule_expression="rate(24 hours)"
)

event_bridge_schedule_arn = schedule(
    pipeline_name=pipeline_name, 
    schedule_expression="cron(0 0-23/1 ? * * 2023-2024)"
)

schedule API 中日期和时间输入的默认时区采用 UTC。有关 EventBridge 调度程序计划表达式的更多信息，请参阅 EventBridge 调度器 API 参考文档ScheduleExpression中的。

计划的特征处理器管道执行为您的转换函数提供了计划执行时间，可用作基于日期范围的输入的幂等性令牌或固定参考点。要禁用（即暂停）或重新启用计划，请分别使用带有 ‘DISABLED’ 或 ‘ENABLED’ 的 schedule API 的 state 参数。

有关特征处理器的信息，请参阅特征处理器 SDK 数据源。

基于事件的执行

可以将特征处理管道配置为在 AWS 事件发生时自动执行。特征处理 SDK 提供了一个接受源事件列表和目标管道的 put_trigger 函数。源事件必须是 FeatureProcessorPipelineEvent（用于指定管道和执行状态事件）的实例。

该put_trigger函数配置 Amazon EventBridge 规则和目标以路由事件，并允许您指定 EventBridge 事件模式以响应任何 AWS 事件。有关这些概念的信息，请参阅 Amazon EventBridge 规则、目标和事件模式。

可以启用或禁用触发器。 EventBridge 将使用 put_trigger API role_arn 参数中提供的角色启动目标管道执行。如果在 Amazon SageMaker Studio Classic 或笔记本环境中使用软件开发工具包，则默认使用执行角色。有关如何获取执行角色的信息，请参阅获取执行角色。

以下示例将：

from sagemaker.feature_store.feature_processor import put_trigger, to_pipeline, FeatureProcessorPipelineEvent

to_pipeline(pipeline_name="target-pipeline", step=transform)

put_trigger(
    source_pipeline_events=[
        FeatureProcessorPipelineEvent(
            pipeline_name="source-pipeline",
            status=["Succeeded"]
        )
    ],
    target_pipeline="target-pipeline"
)

有关使用基于事件的触发器为特征处理器管道创建连续执行和自动重试的示例，请参阅使用基于事件的触发器进行连续执行和自动重试。

有关使用基于事件的触发器创建连续流式处理 和使用基于事件的触发器自动重试的示例，请参阅流式处理自定义数据源示例。