优化 Flux 查询

优化您的 Flux 查询，以减少其内存和计算 (CPU) 需求。

• 使用下推函数启动查询
  • 避免内联处理过滤器
• 避免短窗口持续时间
• 谨慎使用“重”函数
• 尽可能使用 set() 而不是 map()
• 平衡时间范围和数据精度
• 使用 Flux Profiler 衡量查询性能

使用下推函数启动查询

下推函数 是将数据操作推送到底层数据源而不是在内存中操作数据的函数或函数组合。使用下推函数启动查询以提高查询性能。一旦非下推函数运行，Flux 会将数据拉入内存并在那里运行所有后续操作。

下推函数和函数组合

查询 InfluxDB 2.7 或 InfluxDB Cloud 数据源时，大多数下推函数都受支持。如下表所示，InfluxDB 2.7 中不支持少数下推函数。

在查询的开头使用下推函数和函数组合。一旦非下推函数运行，Flux 会将数据拉入内存并在那里运行所有后续操作。

正在使用的下推函数

from(bucket: "example-bucket")
    |> range(start: -1h)                       //
    |> filter(fn: (r) => r.sensor == "abc123") //
    |> group(columns: ["_field", "host"])      // Pushed to the data source
    |> aggregateWindow(every: 5m, fn: max)     //
    |> filter(fn: (r) => r._value >= 90.0)     //

    |> top(n: 10)                              // Run in memory

避免内联处理过滤器

避免内联使用数学运算或字符串操作来定义数据过滤器。内联处理过滤器值会阻止 filter() 将其操作下推到基础数据源，因此先前函数返回的数据会加载到内存中。这通常会导致显着的性能下降。

例如，以下查询使用 仪表盘变量 和字符串连接来定义要过滤的区域。由于 filter() 内联使用字符串连接，因此无法将其操作下推到基础数据源，并将从 range() 返回的所有数据加载到内存中。

from(bucket: "example-bucket")
    |> range(start: -1h)                      
    |> filter(fn: (r) => r.region == v.provider + v.region)

要动态设置过滤器并保持 filter() 函数的下推能力，请使用变量在 filter() 外部定义过滤器值

region = v.provider + v.region

from(bucket: "example-bucket")
    |> range(start: -1h)                      
    |> filter(fn: (r) => r.region == region)

避免短窗口持续时间

窗口化（基于时间间隔对数据进行分组）通常用于聚合和降采样数据。避免短窗口持续时间以提高性能。更多窗口需要更多计算能力来评估每行应分配给哪个窗口。合理的窗口持续时间取决于查询的总时间范围。

谨慎使用“重”函数

以下函数比其他函数使用更多的内存或 CPU。在使用它们之前，请考虑它们在数据处理中的必要性

• map()
• reduce()
• join()
• union()
• pivot()

尽可能使用 set() 而不是 map()

set()、 experimental.set() 和 map 都可以设置数据中的列值，但是 set 函数比 map() 具有性能优势。

使用以下准则来确定使用哪个函数

• 如果将列值设置为预定义的静态值，请使用 set() 或 experimental.set()。
• 如果使用现有行数据动态设置列值，请使用 map()。

将列值设置为静态值

以下查询在功能上是相同的，但是使用 set() 比使用 map() 性能更高。

data
    |> map(fn: (r) => ({ r with foo: "bar" }))

// Recommended
data
    |> set(key: "foo", value: "bar")

使用现有行数据动态设置列值

data
    |> map(fn: (r) => ({ r with foo: r.bar }))

平衡时间范围和数据精度

为确保查询性能，请平衡时间范围和数据精度。例如，如果您查询每秒存储的数据并请求六个月的数据，则结果将包括每个序列 ≈1550 万个点。根据 filter() 后返回的序列数（基数），这可能很快就会变成数十亿个点。Flux 必须将这些点存储在内存中才能生成响应。使用 下推函数 来优化内存中存储的点数。

要查询较长时间段的数据，请创建一个任务来 降采样数据，然后查询降采样后的数据。

使用 Flux Profiler 衡量查询性能

使用 Flux Profiler 包 来衡量查询性能，并将性能指标附加到您的查询输出。以下 Flux Profiler 可用

• query：提供有关整个 Flux 脚本执行的统计信息。
• operator：提供有关查询中每个操作的统计信息。

导入 profiler 包，并使用 profile.enabledProfilers 选项启用 Profiler。

import "profiler"

option profiler.enabledProfilers = ["query", "operator"]

// Query to profile

有关 Flux Profiler 的更多信息，请参阅 Flux Profiler 包。