Flux 1.0 展望

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

导航至

(更新： InfluxDB 3.0 移除了 Flux 和内置任务引擎。用户可以使用外部工具，例如基于 Python 的 Quix，在 InfluxDB 3.0 中创建任务。)

本周在 InfluxDays 大会上，我们宣布 Flux 1.0 即将发布。Flux lang Flux 1.0 版本承诺不再对 Flux 语言进行破坏性更改。重要的是，今天的 Flux 脚本将在 Flux 1.0 上工作，并且从现在到 Flux 1.0 发布之间不会引入任何破坏性更改。随着 1.0 版本的发布，我们还将很快发布一些功能。以下是我们即将推出的功能以及您可能希望利用它们的原因的简要说明

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

  • 版本： 选择加入 Flux 的新功能，而无需担心破坏性更改的风险。

  • 标签多态性： 编写可以抽象数据架构的 Flux 代码。

  • 动态类型： 处理完全动态且无限制的 JSON 数据。

动态类型功能现已可用，您可以在云端试用。现在让我们更详细地了解其中一些功能。

如何使用模块

Flux 模块是一个系统，允许您将 Flux 代码发布到您组织私有的注册表中。然后，您可以从 任务 到 仪表板单元格 以及在您的应用程序的 Flux 代码中，在任何地方导入和重用该代码。

Dynamic Type

图示 Flux 模块和代码共享

这是一个模块发布后如何工作的示例。我们看到的一个常见用例是，组织会使用一个值标记其数据，该值实际上代表的数据远多于标签本身中存在的数据。例如，您可能在数据上有一个 env 标签，表示指标的来源数据中心。该环境的属性在其他标签中不存在，例如哪个云提供商托管该环境，或者该环境是否是生产环境。使用模块，您可以将此信息编码到 Flux 字典中，然后利用它来使用该额外信息注释数据。

使用模块的第一步是定义一个 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 的另一种新类型，它允许使用 Flux 数据类型准确表示任何 JSON 数据。Flux 是一种静态类型语言，它要求容器类型（即数组、记录和字典）包含类型一致的数据。例如，整数列表是一种一致的类型，但字符串和整数列表不是。JSON 没有这些限制，借助新的动态类型功能，我们可以在 Flux 中正确表示 JSON。Flux 仍然是静态类型的，但我们添加了一种名为“dynamic”的新类型，允许 Flux 处理动态类型数据。

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

{
  "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 的新版本。同时，您可以查看我在 2022 年 InfluxDays 大会上的会议。