View a markdown version of this page

顺序执行器 - AWS Elemental MediaTailor

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

顺序执行器

何时使用

A 按顺序SEQUENTIAL_EXECUTOR运行一系列函数。每个步骤都可以使用之前的步骤的结果。

SEQUENTIAL_EXECUTOR当您的逻辑需要多个相互依赖的结果的步骤时使用。常见用例包括获取身份数据,然后使用它来检索受众细分,运行分类步骤,然后根据结果有条件地调用不同的外部服务,以及从多个数据源构建复杂的广告请求网址。

配置字段

SEQUENTIAL_EXECUTOR函数具有以下字段:

  • 运行时 -表达式语言。将此设置为JSONATA

  • FunctionList— 包含 1 到 10 个步骤的有序列表。每个步骤都指定要运行FunctionId的函数。或者,您可以添加RunCondition表达式来控制该步骤是运行还是跳过。

  • 输出 -定义所有步骤完成后要生成的值。每个条目将一个输出键(例如player_params.envelope)映射到一个表达式,该表达式可以引用序列中任何步骤生成的数据。如果省略,则使用序列中各个函数的所有输出。

  • TimeoutMilliseconds(必填)— 整个序列完成的最长时间。如果序列超过此超时时间,则 MediaTailor 丢弃该序列的所有输出。

有序执行和数据流

MediaTailor 从头到尾运行序列中的每个步骤。每个步骤完成后,其生成的值将合并为一组正在运行的结果。后续步骤可以访问原始会话数据以及先前步骤生成的所有值。

临时数据是在步骤之间传递数据的主要机制。当一个函数写入一个temp.*键时,下一步可以读取该值。先前步骤编写的玩家参数和广告请求字段也可供后续步骤查看。

注意

临时数据接受任何数据类型,包括对象和数组。玩家参数和广告请求字段仅接受字符串、数字、布尔值和空值)。

Per-step 运行条件

序列中的每个步骤都有一个可选RunCondition字段。此字段包含返回true或的表达式false。 MediaTailor 在运行该步骤之前立即计算RunCondition表达式。

如果RunCondition表达式的计算结果为false,则完全 MediaTailor 跳过该步骤并移至下一个步骤。如果省略该RunCondition字段,则该步骤将始终运行。

{ "FunctionId": "retryFetch", "RunCondition": "{%temp.statusCode = 500%}" }

此机制允许您构建条件管道。例如,您可以在步骤 1 中运行身份提取,然后仅在步骤 1 返回有效身份时才有条件地在步骤 2 中运行区段查询。

输出模块的工作原理

a 上的输出模块SEQUENTIAL_EXECUTOR控制序列在所有步骤完成后生成的内容:

  • 输出模块存在 — MediaTailor 根据最终累积状态评估输出模块中的表达式,并仅保存这些输出。先前步骤生成的未在顺序输出模块中引用的任何输出都将被丢弃。

  • 输出模块不存在 — 直接 MediaTailor 保存所有步骤的所有累积输出。

提示

当你想让每个函数的输出都通过时,可以省略输出块。当需要在保存累积结果之前过滤、重命名或转换时,请添加输出块。

超时配置

TimeoutMilliseconds字段为整个序列设置了截止日期。此超时涵盖所有步骤,包括函数发出的任何 HTTP 调用。如果序列超过超时时间,则 MediaTailor 丢弃该序列的所有输出,并像没有附加任何函数一样继续操作。

各个HTTP_REQUEST功能仍然尊重自己的RequestTimeoutMilliseconds设置。序列超时充当外部边界,限制总执行时间。

示例:在 HTTP 失败时重试

此示例调用身份 API,并在第一次调用返回服务器错误时自动重试。它使用由 SEQUENTIAL_EXECUTOR 编排的两个 HTTP_REQUEST 函数。

第 1 步 — 主要 fetch (fetchIdentity):

{ "FunctionId": "fetchIdentity", "FunctionType": "HTTP_REQUEST", "HttpRequestConfiguration": { "Runtime": "JSONATA", "MethodType": "GET", "Url": "{%'https://identity.example.com/v1/resolve?ip=' & session.client_ip%}", "Headers": { "Accept": "application/json" }, "RequestTimeoutMilliseconds": 1000, "Output": { "temp.statusCode": "{%response.statusCode%}", "temp.envelope": "{%response.statusCode = 200 ? response.body.envelope : null%}" } } }

第 2 步 — 失败时重试 (retryIdentity):

{ "FunctionId": "retryIdentity", "FunctionType": "HTTP_REQUEST", "HttpRequestConfiguration": { "Runtime": "JSONATA", "MethodType": "GET", "Url": "{%'https://identity-fallback.example.com/v1/resolve?ip=' & session.client_ip%}", "Headers": { "Accept": "application/json" }, "RequestTimeoutMilliseconds": 1000, "Output": { "temp.statusCode": "{%response.statusCode%}", "temp.envelope": "{%response.statusCode = 200 ? response.body.envelope : null%}" } } }

序列 (identityWithRetry):

{ "FunctionId": "identityWithRetry", "FunctionType": "SEQUENTIAL_EXECUTOR", "SequentialExecutorConfiguration": { "Runtime": "JSONATA", "TimeoutMilliseconds": 2000, "FunctionList": [ { "FunctionId": "fetchIdentity" }, { "FunctionId": "retryIdentity", "RunCondition": "{%temp.statusCode >= 500%}" } ], "Output": { "player_params.envelope": "{%temp.envelope%}" } } }

工作原理:

  1. fetchIdentity调用身份 API 并将状态码和信封写入temp.*

  2. 如果状态码为 500 或更高,则RunCondition步骤 2 的计算结果为trueretryIdentity运行。它会temp.envelope用重试响应覆盖temp.statusCode和。

  3. 如果第一次调用成功,则跳过步骤 2。

  4. 输出块写入temp.envelope的序列player_params.envelope