Flux 1.0期待您期待的内容

作者：Nathaniel Cook / 产品
2022年11月03日

(更新: InfluxDB 3.0已放弃使用Flux和内置任务引擎。用户可以使用基于Python的Quix等外部工具在InfluxDB 3.0中创建任务。)

• 模块：在您的组织内部共享Flux代码，并在常见的查询上构建有用的抽象。

• 版本：在新功能中启用Flux，而无需担心破坏性更改的风险。

• 标签多态性：编写可以抽象数据模式（schema）的Flux代码。

• 动态类型：与完全动态且无限制的JSON数据进行交互。

动态类型功能现已可用，您可以在云中试用。现在让我们更详细地探讨一些这些功能。

如何使用模块

使用模块的第一步是定义一个Flux脚本

// Notice we give this script a package name since it will be part of a module.
package env

import "dict"

// Map envs to their cloud provider.
providers =
	[
    	"prod01": "aws",
    	"prod02": "aws",
    	"prod03": "gcp",
    	"prod04": "azure",
    	"stag01": "aws",
    	"stag02": "gcp",
	]

// Map envs to their purpose.
// We could just regex match on the name,
// but sometimes it’s important to be explicit.
kinds =
	[
    	"prod01": "production",
    	"prod02": "production",
    	"prod03": "production",
    	"prod04": "production",
    	"stag01": "staging",
    	"stag02": "staging",
	]

// annotate adds extra information about the env to the table.
//
// ## Parameters
// - tables: Input data. Default is piped-forward data (`<-`).
//
annotate = (tables=<-) =>
	tables
    	|> map(
        	fn: (r) =>
            	({r with provider: dict.get(dict: providers, key: r.env, default: "unknown"),
                	kind: dict.get(dict: kinds, key: r.env, default: "unknown"),
            	}),
    	)

然后我们使用API发布此模块。有关发布过程的更多细节将在模块发布时提供。发布后，我们可以在仪表板单元格中导入此模块，并按云提供商对数据进行分组，并过滤出预发布环境。

import "modules/env"

// Render median request duration grouped by provider in only production envs
from(bucket: "request")
	|> range(start: v.timeRangeStart, stop: v.timeRangeStop)
	|> filter(fn: (r) => r._measurement == "duration")
	|> env.annotate()
	|> filter(fn: (r) => r.kind == "production")
	|> group(columns: ["provider"])
	|> aggregateWindow(every: v.windowPeriod, fn: median)

版本和重大更改

版本是Flux 1.0发布的关键组成部分。在1.0版本中，我们将不再对Flux进行重大更改。然而，偶尔有一些我们应添加到Flux中的有价值功能，这将构成重大更改。我们需要一种安全地添加这些功能而不破坏Flux脚本的方法。这就是版本的作用。版本是一组默认禁用的功能。您可以选择加入一个版本，这将启用这些功能，但如果您不选择加入，则不会发生任何变化，您的脚本将继续正常运行而不会破坏。由于版本是可选的，您可以在更新到新版本之前更新您的脚本，以确保它们按预期继续工作。可以按脚本逐个选择版本，或通过API或组织级别设置启用，从而在升级脚本方面提供灵活性。

创建新版本将是一个罕见的事件，我们预计每年最多创建一个新版本。因此，我们将版本命名为它们创建的年份。Flux的第一个版本将称为 2022.1。我们可能将在明年发布Flux的第二版，并将其命名为 2023.1。

用户可以通过在脚本中指定它来启用新版本

// Enable a specific edition with this line as the first line in the script
#edition("2023.1")

// The rest of the script follows

标签多态性预期效果

标签多态性是我们将在版本之后发布的第一批功能之一，因为该功能为Flux添加了新的语法，这是破坏性的。

标签多态性允许用户抽象其数据模式。您是否曾经想要编写一个Flux函数，该函数接受列名作为参数，然后在函数中使用该名称动态地？这一直是Flux的一个长期请求的功能，它即将作为Flux第二版的一部分到来。

让我们看看我们之前使用的显示模块的相同示例。我们在env包中编写了这个函数

// annotate adds extra information about the env to the table.
//
// ## Parameters
// - tables: Input data. Default is piped-forward data (`<-`).
//
annotate = (tables=<-) =>
	tables
    	|> map(
        	fn: (r) =>
            	({r with provider: dict.get(dict: providers, key: r.env, default: "unknown"),
                	kind: dict.get(dict: kinds, key: r.env, default: "unknown"),
            	}),
    	)

如果我们仔细观察这个函数，我们会意识到它硬编码了关于数据模式的大量信息。它假设可以像r.env一样访问env标记，并且还硬编码了输出表上的列provider和kind。如果我们有来自不一致系统（有时使用env，有时使用environment）的数据怎么办？我们可以使用labels来抽象我们正在使用的列，如下所示

// We need to enable the edition in order to use labels
#edition("2023.1")

// annotate adds extra information about the env to the table.
//
// ## Parameters
// - env: Label for the environment column. Default `env`.
// - provider: Output label for the provider column. Default `provider`.
// - kind: Output label for the kind column. Default `kind`.
// - tables: Input data. Default is piped-forward data (`<-`).
//
annotate = (tables=<-, env=.env, provider=.provider, kind=.kind) =>
	tables
    	|> map(
        	fn: (r) =>
            	({r with [provider]: dict.get(dict: providers, key: r[env], default: "unknown"),
                	[kind]: dict.get(dict: kinds, key: r[env], default: "unknown"),
            	}),
    	)

这里有三个新的语法元素。首先，我们有.env，它定义了一个值为env的标签，以及.，它告诉Flux这是一个标签。我们使用标签值作为我们添加到函数的三个列参数的默认值。其次，我们看到r[env]。这是我们访问记录r上的env列的方式。最后，在输出记录中，我们看到[provider]和[kind]，它们基于provider和kind标签的值在输出记录上添加列。

以下是在使用遗留系统时调用此新函数的示例，该系统使用 environment 而不是 env，并且已经有一个名为 kind 的标签。

// We need to enable the edition in order to use labels
#edition("2023.1")

import "modules/env"

// Render median request duration grouped by provider in all production envs
from(bucket: "legacy_request")
	|> range(start: v.timeRangeStart, stop: v.timeRangeStop)
	|> filter(fn: (r) => r._measurement == "duration")
	// We can specify the column names when calling the annotate function
	|> env.annotate(env: .environment, kind: .env_kind)
	|> filter(fn: (r) => r.env_kind == "production")
	|> group(columns: ["provider"])
	|> aggregateWindow(every: v.windowPeriod, fn: median)

标签和版本 2023.1 尚未发布，因此上述脚本在现有的 Flux 上将报错，但请关注这些即将推出的功能。

JSON 数据的动态类型

动态类型是 Flux 新增的一种类型，允许任何 JSON 数据以 Flux 数据类型被准确表示。Flux 是一种静态类型语言，它要求容器类型（例如数组、记录和字典）包含相同类型的数据。例如，整数列表是相同类型，但字符串和整数的列表则不是。JSON 没有这些限制，而通过新的动态类型功能，我们可以在 Flux 中正确表示 JSON。Flux 仍然是静态类型，但我们增加了一种名为“动态”的新类型，允许 Flux 处理动态类型的数据。

例如，此前的 Flux 中无法表示此 JSON

{
  "items":  [
  	{"name": "a", "pos": {"x": 10, "y": 10}},
  	{"name": "b", "pos": {"x": 30}},
  	{"name": "c", "pos": {}},
  	{"name": "d"}
	]
}

与其他功能不同，动态类型现在即可使用。

现在您可以使用新的 experimental/dynamic 包解析上述 JSON 而不会出错。

import "experimental/dynamic"

parsed = dynamic.parseJSON(data: jsonDataAsBytes)

这是可能的，因为函数 dynamic.parseJSON 返回一个类型为 dynamic 的 Flux 值。

动态值表示在运行时才知道类型的值。Flux 还可以有动态值的容器（例如数组等）。

动态值不允许在 Flux 表中，因此我们显式地将动态值转换为具有静态已知类型的值。此转换使用了 array 包的辅助函数和 dynamic 包中的几个新函数。

import "array"
import "experimental/dynamic"

parsed = dynamic.parseJSON(data: jsonDataAsBytes)

// parsed is dynamic so parsed.items is also dynamic
parsed.items
	// convert the plain dynamic value to an array of dynamic values
	|> dynamic.asArray()
	// map over each dynamic value in the array and construct a record
	|> array.map(fn: (r) => ({
    	name: string(v: x.name),
    	posX: int(v: x.pos.x),
    	posY: int(v: x.pos.y),
	}))
	// convert the array of records (no longer containing any dynamic values) into a table
	|> array.from()

对于没有 pos 值的 JSON 行，表中将包含 nulls。

结语

Flux 1.0 包含了很多内容。许多新功能即将推出，动态类型现在即可使用。关注 Flux 1.0 的发布和新版 Flux 的推出。在此期间，您可以查看我在 InfluxDays 2022 上的会议。