使用新的 Flux "types" 包

作者：Sean Brickley / 用例, 开发者, 产品
2022 年 3 月 7 日

作为一种强类型语言，Flux 可以保护您免受许多潜在的运行时故障的影响。但是，如果您不知道正在查询的数据的列类型，您可能会遇到一些烦人的错误。

假设您有一个存储桶，它接收来自多个不同流的定期写入，并且您想要编写一个任务，将该存储桶中的一个 measurement 进行降采样到另一个存储桶中。如果您预先知道，例如，_value 列将始终是数字类型，则可以毫无问题地运行以下任务

option task = {name: "write-agg", every: 30m, offset: 1s}

from(bucket: "test-bucket")
       |> range(start: -30m)
       |> filter(fn: (r) => r._measurement == "logs")
       |> aggregateWindow(fn: mean, every: 5m)
       |> to(bucket: "test-downsample")

但是，如果您在查询之前不知道数据的模式，则可能会遇到麻烦。例如，如果 _value 最终是一个字符串而不是数字，则此查询可能会失败。

到目前为止，对于这个问题还没有通用的解决方案。如果您知道要查找的标签，则对 _field 进行过滤可能会奏效，但也许您不知道这些详细信息，或者您需要包含或排除的标签列表太长，无法舒适地放入过滤器谓词中。

输入 types 包。此包引入了 isType 函数，这使得按列类型进行过滤变得更加容易。我们可以通过导入 types 包并添加一个新的过滤器来检查 r._value 的类型，从而修复我们的原始查询。

import "types"

option task = {name: "write-agg", every: 30m, offset: 1s}

from(bucket: "test-bucket")
       |> range(start: -30m)
       |> filter(fn: (r) => r._measurement == "logs")
       |> filter(fn: (r) => types.isType(v: r._value, type: "float"))
       |> aggregateWindow(fn: mean, every: 5m)
       |> to(bucket: "test-downsample")

现在我们可以确定，任何管道输入到 aggregateWindow 的数据都具有 _value 列，其类型为 float，从而避免任何潜在的类型错误。果然，我们的任务成功完成，并将降采样数据成功写入了我们的新存储桶

我们还可以使用 isType 进行更复杂的过滤。让我们想象一下，我们从上面任务中读取的 logs measurement 具有许多不同类型的字段。我们想要聚合所有这些字段，但我们知道某些聚合不适用于每种类型。我们可以使用 isType 来根据我们找到的数据类型决定使用哪个聚合函数。

import "types"

option task = {name: "write-agg", every: 30m, offset: 1s}

from(bucket: "test-bucket")
       |> range(start: -30m)
       |> filter(fn: (r) => r._measurement == "logs")
       |> filter(fn: (r) => {
                 return types.isType(v: r._value, type: "float")
                         or types.isType(v: r._value, type: "int")
                         or types.isType(v: r._value, type: "uint")
       })	
       |> aggregateWindow(fn: mean, every: 5m)
       |> to(bucket: "test-downsample")

from(bucket: "test-bucket")
       |> range(start: -30m)
       |> filter(fn: (r) => r._measurement == "logs")
       |> filter(fn: (r) => {
                return types.isType(v: r._value, type: "string")
                        or types.isType(v: r._value, type: "bool")
       })
      |> aggregateWindow(fn: last, every: 5m)
      |> to(bucket: "test-downsample")

这个新包包含在最新版本的 Flux 中，并且对所有云用户都可用。我们鼓励您试用并告诉我们您的想法！