TL;DR InfluxDB 技术提示：使用 Flux 将 IoT 数据从边缘传输到云端

作者：Anais Dotis-Georgiou / 产品, 用例, 开发者
2021 年 12 月 13 日

导航至


在将数据写入 InfluxDB 时，您有很多选择。您可以

  • 将数据写入边缘或雾设备的 OSS 实例
  • 将数据从您的 IoT 设备直接写入您的 InfluxDB Cloud
  • 将数据写入边缘设备上 InfluxDB 的 OSS 实例，并将数据临时推送到 InfluxDB Cloud

最后一个选项是维护和管理 IoT 设备群的最强大和最灵活的方式。该架构为您提供以下几个优势，包括

  • 能够执行任务，在将数据发送到 InfluxDB Cloud 之前准备和清理数据。这不仅可以帮助您创建组织良好的数据管道，还可以确保您的数据已在 IoT 设备之间标准化
  • 灵活地仅将选定的数据发送到 InfluxDB Cloud，例如在 OSS 实例中标记的异常
  • 隔离任务工作负载并使其靠近数据源的选项
  • 在您的边缘设备、InfluxDB Cloud 实例和终端客户之间增加一层安全性的好处

最后一个选项的架构图。传感器将数据写入边缘的 InfluxDB 的 OSS 实例，然后该实例将数据写入 InfluxDB Cloud。将数据从边缘写入 InfluxDB 云端

使用 to() 函数将数据从 OSS 合并到 InfluxDB Cloud

实现此架构的一种方法是使用 to() 函数。to() 函数现在支持从 OSS v2.1 实例将数据写入 InfluxDB Cloud 实例的选项。

Using the to() function in Flux - an example

使用 to() 函数将数据从 OSS 实例写入 InfluxDB Cloud 的示例

在上述仪表板的右侧，用户正在查询 OSS 实例。在右侧，InfluxDB Cloud 实例在数据通过 to() 函数写入 InfluxDB Cloud 后，显示 OSS 实例中 Flux 脚本的结果。

要使用 to() 函数从 InfluxDB Cloud 写入数据

  1. 查询您的数据。
from(bucket: "RoomSensors")
|> range(start: -10m)
|> filter(fn: (r) => r["_measurement"] == "temperature")
|> filter(fn: (r) => r["_field"] == "temperature")
|> aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: false)
  1. 指定目标主机 URL、orgID、令牌和存储桶 ID。
|> to(host: https://us-west-2-1.aws.cloud2.influxdata.com, 
        orgID:   04xxx00,
        token: 0xxx==, 
        bucketID: 43xxx00)

您首先需要拥有一个 InfluxDB Cloud 帐户才能试用。免费套餐帐户也可以正常工作。您可以轻松地在此处注册帐户。注册后，您需要

to() 函数的局限性和替代方案

to() 函数提供了一种将数据从边缘设备写入云端的简便方法。但是，它有一些局限性

  • 通过 HTTP 发送
  • 没有内置的故障处理功能
  • 没有内置的批量处理、重试或并行化功能

如果满足以下条件，则 to() 函数应仅用于将数据从 OSS 合并到云端

  • 您打算在将数据写入云端之前先对其进行降采样，以限制请求的大小。
  • 您拥有少量设备，或者以这种方式写入相对少量的数据。您不能将 to() 函数用于大型工作负载，因为数据量可能会导致网络故障。也没有内置的故障处理功能。
  • 您不是试图通过微批量处理和生成非常高的请求计数来克服写入大量数据的问题。

虽然 to() 函数有一些局限性，但还有另一种选择可用于将 IoT 数据从 OSS 写入 InfluxDB Cloud。如果您想处理更大的工作负载，您可以先使用 mqtt.to() 函数将数据写入 MQTT 代理。

在降采样任务中可接受的 to() 用法

现在我们了解了 to() 函数的局限性，让我们重点介绍一个可能成功的示例。对于此示例，我们假设我们想要使用任务按计划降采样和写入数据。

option task = {
    name: "downsample and consolidate task",
    every: 1d,
    offset: 10m,
}
import "http"
import "influxdata/influxdb/secrets"

myToken = "Token ${secrets.get(key: "cloud-token")}"
myUrl= "https://us-west-2-1.aws.cloud2.influxdata.com"
myOrgID = "Token ${secrets.get(key: "cloud-org")}"
myBucketID = "Token ${secrets.get(key: "cloud-bucket")}"

from(bucket: "sensorData")
|> range(start: -task.every)
|> filter(fn: (r) =>
(r["_measurement"] == "sensors"))
|> mean()
|> to(host: myurl, 
        orgID:   myOrgID,
        token: myToken, 
        bucketID: myBucketID)

此任务将一个指标中的所有数据聚合为一个每日平均值，然后再将其写入 InfluxDB Cloud。

延伸阅读

虽然这篇文章旨在全面概述 to() 函数，但以下资源也可能让您感兴趣

  1. 如何将 OSS 数据合并到云帐户：这篇文章描述了如何使用 http.to() 函数而不是 to() 函数将来自多个 OSS 实例的数据共享到 InfluxDB Cloud。此博客基本上是您当前阅读的博客的前身。
  2. InfluxDB 任务入门：本文档介绍了如何开始使用 InfluxDB 任务，以防您以前从未创建过任务。
  3. TL;DR InfluxDB 技术提示 – 使用任务和检查进行 InfluxDB 监控：这篇文章是一个关于如何使用 UI 创建任务以及检查（一种特殊类型的任务）的教程。
  4. 编写任务和设置 InfluxDB Cloud 警报：本教程演示了如何为那些不习惯使用 UI 的人使用 VS Code 的 Flux 插件创建任务。
  5. 使用 Flux VS Code 扩展进行 IoT 应用程序开发。这篇文章重点介绍了 Flux 插件 for VS Code 的一些新功能，这些功能允许您轻松地将 Flux 查询转换为任务。
  6. 通过 InfluxDB 降采样来趋势化聚合值：本教程重点介绍了如何使用任务来执行各种聚合。

关于 InfluxDB 的 to() 函数的最终想法

我希望这篇 InfluxDB 技术提示文章能够启发您利用 to() 函数将数据从边缘设备合并到 InfluxDB Cloud。如果您正在使用 Flux 并需要帮助，请在我们的社区站点或 Slack 频道中提问。如果您正在 InfluxDB 之上开发一个很酷的 IoT 应用程序，我们很乐意听到您的故事，因此请务必分享您的故事！此外，请在评论区分享您的想法、疑虑或问题。我们很乐意获得您的反馈并帮助您解决遇到的任何问题！