Watcher 转换上下文
编辑Watcher 转换上下文编辑
使用 Painless 脚本作为 watch 转换 将有效负载转换为新的有效负载,以便在 watch 中进一步使用。转换脚本返回新有效负载的 Object 值。
以下变量在所有 watcher 上下文中都可用。
变量
-
params
(Map
,只读) - 作为查询的一部分传入的用户定义参数。
-
ctx['watch_id']
(String
,只读) - watch 的 ID。
-
ctx['id']
(String
,只读) - 服务器为运行的 watch 生成的唯一标识符。
-
ctx['metadata']
(Map
,只读) - 元数据可以添加到 watch 定义的顶层。这是用户定义的,通常用于合并 watch 中的重复值。
-
ctx['execution_time']
(ZonedDateTime
,只读) - watch 开始执行的时间。
-
ctx['trigger']['scheduled_time']
(ZonedDateTime
,只读) - watch 的计划触发时间。这是 watch 应该执行的时间。
-
ctx['trigger']['triggered_time']
(ZonedDateTime
,只读) - watch 的实际触发时间。这是触发 watch 执行的时间。
-
ctx['payload']
(Map
,只读) - 基于 watch 输入 的可访问 watch 数据。
返回值
-
Object
- 新的有效负载。
API
可以使用标准的 Painless API。
示例
要运行示例,请先按照 上下文示例 中的步骤操作。
POST _watcher/watch/_execute { "watch" : { "trigger" : { "schedule" : { "interval" : "24h" } }, "input" : { "search" : { "request" : { "indices" : [ "seats" ], "body" : { "query" : { "term": { "sold": "true"} }, "aggs" : { "theatres" : { "terms" : { "field" : "play" }, "aggs" : { "money" : { "sum": { "field" : "cost" } } } } } } } } }, "transform" : { "script": """ return [ 'money_makers': ctx.payload.aggregations.theatres.buckets.stream() .filter(t -> { return t.money.value > 50000 }) .map(t -> { return ['play': t.key, 'total_value': t.money.value ] }).collect(Collectors.toList()), 'duds' : ctx.payload.aggregations.theatres.buckets.stream() .filter(t -> { return t.money.value < 15000 }) .map(t -> { return ['play': t.key, 'total_value': t.money.value ] }).collect(Collectors.toList()) ] """ }, "actions" : { "my_log" : { "logging" : { "text" : "The output of the payload was transformed to {{ctx.payload}}" } } } } }
Java Stream API 用于转换中。此 API 允许在管道中操作列表的元素。 |
|
流过滤器会删除不符合过滤器条件的项目。 |
|
流映射将每个元素转换为一个新对象。 |
|
收集器将流简化为 |
|
对转换中的第二组值再次执行此操作。 |
以下操作转换将 mod_log 操作中的每个值更改为 String
。此转换不会更改 unmod_log 操作中的值。
POST _watcher/watch/_execute { "watch" : { "trigger" : { "schedule" : { "interval" : "24h" } }, "input" : { "search" : { "request" : { "indices" : [ "seats" ], "body" : { "query" : { "term": { "sold": "true"} }, "aggs" : { "theatres" : { "terms" : { "field" : "play" }, "aggs" : { "money" : { "sum": { "field" : "cost" } } } } } } } } }, "actions" : { "mod_log" : { "transform": { "script" : """ def formatter = NumberFormat.getCurrencyInstance(); return [ 'msg': ctx.payload.aggregations.theatres.buckets.stream() .map(t-> formatter.format(t.money.value) + ' for the play ' + t.key) .collect(Collectors.joining(", ")) ] """ }, "logging" : { "text" : "The output of the payload was transformed to: {{ctx.payload.msg}}" } }, "unmod_log" : { "logging" : { "text" : "The output of the payload was not transformed and this value should not exist: {{ctx.payload.msg}}" } } } } }
此示例以非常相似的方式使用流式 API。以下差异很细微,但值得一提。
以下示例显示了在完整 watch 的上下文中使用脚本化的 watch 和操作转换。此 watch 还使用脚本化的 条件。
POST _watcher/watch/_execute { "watch" : { "metadata" : { "high_threshold": 4000, "low_threshold": 1000 }, "trigger" : { "schedule" : { "interval" : "24h" } }, "input" : { "search" : { "request" : { "indices" : [ "seats" ], "body" : { "query" : { "term": { "sold": "true"} }, "aggs" : { "theatres" : { "terms" : { "field" : "play" }, "aggs" : { "money" : { "sum": { "field" : "cost", "script": { "source": "doc.cost.value * doc.number.value" } } } } } } } } } }, "condition" : { "script" : """ return ctx.payload.aggregations.theatres.buckets.stream() .anyMatch(theatre -> theatre.money.value < ctx.metadata.low_threshold || theatre.money.value > ctx.metadata.high_threshold) """ }, "transform" : { "script": """ return [ 'money_makers': ctx.payload.aggregations.theatres.buckets.stream() .filter(t -> { return t.money.value > ctx.metadata.high_threshold }) .map(t -> { return ['play': t.key, 'total_value': t.money.value ] }).collect(Collectors.toList()), 'duds' : ctx.payload.aggregations.theatres.buckets.stream() .filter(t -> { return t.money.value < ctx.metadata.low_threshold }) .map(t -> { return ['play': t.key, 'total_value': t.money.value ] }).collect(Collectors.toList()) ] """ }, "actions" : { "log_money_makers" : { "condition": { "script" : "return ctx.payload.money_makers.size() > 0" }, "transform": { "script" : """ def formatter = NumberFormat.getCurrencyInstance(); return [ 'plays_value': ctx.payload.money_makers.stream() .map(t-> formatter.format(t.total_value) + ' for the play ' + t.play) .collect(Collectors.joining(", ")) ] """ }, "logging" : { "text" : "The following plays contain the highest grossing total income: {{ctx.payload.plays_value}}" } }, "log_duds" : { "condition": { "script" : "return ctx.payload.duds.size() > 0" }, "transform": { "script" : """ def formatter = NumberFormat.getCurrencyInstance(); return [ 'plays_value': ctx.payload.duds.stream() .map(t-> formatter.format(t.total_value) + ' for the play ' + t.play) .collect(Collectors.joining(", ")) ] """ }, "logging" : { "text" : "The following plays need more advertising due to their low total income: {{ctx.payload.plays_value}}" } } } } }
以下示例显示了元数据的用法以及将日期转换为可读格式。
POST _watcher/watch/_execute { "watch" : { "metadata" : { "min_hits": 10 }, "trigger" : { "schedule" : { "interval" : "24h" } }, "input" : { "search" : { "request" : { "indices" : [ "seats" ], "body" : { "query" : { "term": { "sold": "true"} }, "aggs" : { "theatres" : { "terms" : { "field" : "play" }, "aggs" : { "money" : { "sum": { "field" : "cost" } } } } } } } } }, "condition" : { "script" : """ return ctx.payload.hits.total > ctx.metadata.min_hits """ }, "transform" : { "script" : """ def theDate = ZonedDateTime.ofInstant(ctx.execution_time.toInstant(), ctx.execution_time.getZone()); return ['human_date': DateTimeFormatter.RFC_1123_DATE_TIME.format(theDate), 'aggregations': ctx.payload.aggregations] """ }, "actions" : { "my_log" : { "logging" : { "text" : "The watch was successfully executed on {{ctx.payload.human_date}} and contained {{ctx.payload.aggregations.theatres.buckets.size}} buckets" } } } } }