Flux：使用 InfluxDB 进行边缘数据复制的关键

作者：Anais Dotis-Georgiou / 产品
2022 年 10 月 06 日

导航至

本文最初发表在 The New Stack 上，并在此处获得许可后重新发布。

边缘数据复制 (EDR) 允许开发者充分利用 InfluxDB 在边缘的能力。开发者还可以将相同的数据用于云中的不同目的。

Flux 是 InfluxDB 时间序列数据库平台的脚本和数据查询语言，它提供了诸如 边缘数据复制 (EDR) 等有用功能。

EDR 允许开发者从 InfluxDB 的开源实例中的某个桶自动复制数据到 InfluxDB Cloud 的某个实例。这个功能改变了物联网开发者构建基于 InfluxDB 的 IoT 应用的思维方式，因为它在系统不同元素之间创建了持久和可靠的数据传输。EDR 允许开发者在边缘利用 InfluxDB 的全部功能，同时允许开发者在云中使用相同的数据或数据的子集，用于完全不同的目的。

EDR 可以实现诸如从边缘源收集高保真数据、在边缘清理和转换数据、在云中整合边缘数据以生成对整个边缘的全球视图，以及进行云中整合数据的时序分析和警报等任务。

Flux 的优势

作为一个完全功能的脚本和查询语言，Flux 具有强大的功能。它允许您查询时间序列数据，为创建警报、管理时间序列数据生命周期、以及清理、分析和转换数据打下基础。

Flux 的优势在于它提供在服务器端执行这些查询和功能的能力，从而提高性能。实际上，一些 Flux 函数将查询推送到 InfluxDB 平台，在存储上进行执行，以优先考虑速度和内存优化。

您可以在这里了解更多关于推送的信息。

使用 Flux 查询

一个简单的 Flux 查询看起来像这样

from(bucket: "bucket1") 
|> range(start: -10m)
|> filter(fn: (r) => r._measurement == "measurement1")
|> filter(fn: (r) => r.tag1 == "tagvalue1")
|> filter(fn: (r) => r._field == "field1")

您可以使用 from() 函数选择您要查询数据的存储桶。range() 函数指定您数据查询的时间窗口或间隔。最后，filter() 函数允许您指定您想要查询的数据子集。

InfluxDB 使用度量来将大量数据子集分组或结构化，使用标签存储关于您的时间序列数据的元数据，并使用字段存储您的时间序列数据的实际值。

例如，要查询天气数据，一个 Flux 查询可能看起来像这样

from(bucket: "weather") 
|> range(start: -10m)
|> filter(fn: (r) => r._measurement == "United States")
|> filter(fn: (r) => r.location == "Austin")
|> filter(fn: (r) => r.temperature == 78.0)

要了解更多关于编写 Flux 查询的信息，有很多资源可以帮助您入门。

管理您的时序数据生命周期

Flux 允许您以多种方式转换和汇总时序数据。Flux 的一个常见用途是创建任务来 下采样数据 以创建数据的物化视图。下采样将高分辨率数据转换为低分辨率的摘要。除了其他方面，下采样可以减少 InfluxDB 实例的整体磁盘大小，同时保留数据的整体形状。

Flux 中的下采样任务可能如下所示

// Task Options
option task = {name: "downsample task example", every: 1w}

// Defines a data source
data = from(bucket: "weather")
// queries data for 1w based off of the task configuration options above
|> range(start: -task.every)
|> filter(fn: (r) => r._measurement == "United States")
|> filter(fn: (r) => r.location == "Austin")
|> filter(fn: (r) => r.temperature == 78.0)

data
// Windows and aggregates the data in to 1h averages
|> aggregateWindow(fn: mean, every: 1h)
// Stores the aggregated data in a new bucket
|> to(bucket: "data-downsampled")

此任务将高精度温度数据收集为一小时的平均值，然后使用 to() 函数将数据写入一个新的存储桶。

对您的数据进行警报

鉴于最近 边缘数据复制功能的推出，我们想深入了解 Flux 如何用于启用从边缘到云的数据转换。Flux 提供了多种方式来检查您的数据，以查看它是否满足某些条件，然后对其发出警报。您可以使用 Flux 根据您在边缘使用或收集的数据以及云中汇总的数据设置警报。

在 Flux 中创建警报有几种不同的方法。您可以通过 InfluxDB Cloud UI 构建阈值或“死男人”检查。此方法会自动为每个创建的警报生成 Flux 脚本，并使用 InfluxDB 检查和通知系统。此外，您可以编写自己的自定义检查和通知任务。您如何检查和警报数据取决于您。Flux 在处理数据方面为用户提供了很多灵活性。您可以使用 Flux 将警报发送到以下 通知端点，例如 Slack、Microsoft teams、PagerDuty 以及许多其他地方。

Notification endpoints with Flux

要了解更多关于 创建检查和通知 的信息，请查看以下资源

  • 一个警报任务：本节“快速掌握”将介绍如何编写一个检查数据并发送Slack警报的任务。
  • 检查和通知：本节“快速掌握”详细介绍了InfluxDB中检查和通知系统的工作原理。
  • 在InfluxDB UI中创建检查：本文档介绍了使用InfluxDB和Flux创建检查的几种不同方法。
  • 自定义检查文档：InfluxDB文档总是Flux和InfluxDB相关内容的绝佳资源。

分析、清洁和转换您的数据

对时间序列数据进行转换和统计分析可以从中获得有价值的见解。您可能还需要在使用边缘数据复制功能和将其整合到云中之前清洁和准备您的数据。

Flux拥有许多用于分析和时间序列数据准备的不同函数。有用于统计时间序列分析的转换函数，用于动态统计和基本时间序列分析的转换函数，以及用于财务分析的技术动量指标。此外，还有用于处理地理时间数据的Math包。

Flux还可以转换数据。例如，连接（视频）、联合和旋转（视频）会改变数据的形状。

成功使用时间序列数据的另一个关键组件是操纵时间戳的能力。有一组函数允许您操纵时间戳并执行各种时间序列转换。

您还可以使用Flux编写自定义算法。以下链接展示了使用Flux编写的两个用于异常检测的不同算法

  1. 使用中值绝对偏差进行异常检测
  2. 深入探讨Flux中的机器学习：朴素贝叶斯分类

InfluxDB边缘数据复制的强大功能

最近，InfluxDB发布了一个名为“边缘数据复制”的新功能。EDR允许您在InfluxDB开源实例中配置一个桶，以自动将此桶中的数据复制到InfluxDB云实例中的另一个桶。

例如，使用EDR，您可以在边缘或雾中运行多个OSS实例并收集高精度数据。然后，您可以创建任务，在自动复制到InfluxDB云实例之前对数据进行下采样、处理或转换。您可以使用此功能构建混合架构，通过最小化传输到云的数据量来节省成本，或者在使用云数据存储之前清洁数据，使其更高效。

设置边缘数据复制仅需几个步骤

  1. 使用类似influx remote的命令通过CLI创建远程连接
    influx remote create \
--name example-remote-name \
--remote-url https://us-west-2-1.aws.cloud2.influxdata.com \
--remote-api-token mYsuP3r5Ecr37t0k3n \
--remote-org-id 00xoXXoxXX00
  2. 添加任何所需的下采样或转换任务并将数据写入OSS中的新桶。
  3. 使用类似influx replication的命令从OSS桶创建到InfluxDB云桶的复制流
    influx replication create \
--name example-replication-stream-name \
--remote-id 00xoXXXo0X0x \
--local-bucket-id Xxxo00Xx000o \
--remote-bucket-id 0xXXx00oooXx

总结

希望这篇文章能激发您探索InfluxDB、Flux和边缘数据复制的兴趣。要了解更多关于这些主题的信息，请查看我们在数据+AI峰会会议上即将进行的演讲，我们将详细讨论这些主题，或者在社区Slack频道上与我联系。