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.*键时,下一步可以读取该值。之前步骤中写入的玩家参数和广告请求字段也可供后续步骤查看。

注意

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

Per-step 运行条件

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

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

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

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

输出模块的工作原理

上的输出块SEQUENTIAL_EXECUTOR控制着所有步骤完成后序列产生的结果:

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

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

提示

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

超时配置

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

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

示例:在 HTTP 失败时重试

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

步骤 1-主提取 (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