InfluxDB Cloud物联网原型设计（第二部分）：查询、任务和仪表板

作者：Rick Spencer / 应用案例，产品，开发者
2019年9月27日

导航至

这是四个部分系列的第二部分。请阅读第一部分，第三部分，和第四部分。

物联网场景

之前，我组装了一个物联网设备（本系列的第一部分《使用InfluxDB Cloud原型设计物联网》），并将其数据流传输到我的InfluxDB Cloud账户。从那时起，数据一直快乐地传输到我的账户。

接下来，我想创建一个仪表板，显示土壤湿度随时间的变化，以及当前的土壤湿度读数。

平滑土壤湿度读数

如您所见，土壤湿度读数可能会有一点波动，我想稍微对其进行平滑处理。此外，我的PlantBuddy桶每3秒存储每个传感器的读数。虽然这对某些类型分析和功能非常有用，但长期存储所有这些数据几乎没有价值，并将增加项目的成本。

有多种方法可以解决这个问题。我将采取一个简单的方法进行第一次迭代。

我将创建一个新的桶来存储降采样数据。
我将编写一个查询，从原始桶中提取最后5分钟的数据，计算平均值，并将这些数据放入新的“压缩”桶中。
我将从这个查询创建一个每五分钟运行一次的任务。

创建一个新的桶

首先，我创建一个新的桶来存储降采样数据。请注意，我仍然在使用免费账户，所以我目前还不能将保留时间提高到72小时以上。

使用Flux“压缩”数据

现在，我将在数据探索器中编写一个Flux查询以压缩数据。

Flux简介

如果您不熟悉Flux，Flux是一种新的查询语言，旨在成为一种通用数据处理语言。它设计得既简单易用，又足够强大，可以处理任何数据处理任务。

从概念上讲，Flux让我想起了一些awk的管道操作。基本上，你编写一系列语句，这些语句将一个数据表作为输入，对表进行一些修改，然后输出一个新的表供下一个语句使用。

Flux具有以下有趣的语言特性组合

Flux是强类型和静态类型的。这意味着许多错误可以在创建时就被捕获，而且还可以获得丰富的编辑器功能，例如语句补全、内联帮助等。
Flux在声明变量时不需要指定类型。虽然Flux是强类型和静态类型的，但它能够从上下文中推断类型，所以在编写Flux代码时，你不必担心类型问题。
Flux在所有地方都使用关键字参数。这使得Flux代码更容易阅读和理解。

如果您对以上内容都不理解，或者您根本不在乎，那也没关系。可能通过展示比解释更容易理解。

首先，转到数据探索器视图。

然后点击“脚本编辑器”以切换到脚本编辑器视图。

现在我可以交互式地编写一些Flux。通常，一个Flux脚本将以一个“from”语句开始。这告诉脚本从哪里获取第一个要处理的数据表。在这种情况下，我想从PlantBuddy桶中获取表，因为那里有所有数据。“from”是一个函数，Flux中的函数使用命名参数。因此，我将告诉我的查询从名为“PlantBuddy”的桶中获取初始数据，如下所示

from(bucket: "PlantBuddy")

这将简单地返回桶中所有数据的表。然而，InfluxDB Cloud 2.0不支持这一点，因为这可能是大量的数据导出，并且很可能不是您想要做的。请注意，这并不是Flux语言的限制，而只是为了防止用户意外超出使用配额，并且不会因为意外的查询而过度使用数据库引擎。

因此，有必要添加至少一个管道来限制选择的数据。在 Flux 中，您使用管道向前运算符（|>）将表格输出到查询中的下一个操作。这显然与 Unix 系统中的管道符号类似。所以，首先，我将向 InfluxDB 请求仅从桶中获取最后 5 分钟的数据。我通过将桶中的数据“管道化”到范围函数来实现这一点。范围函数具有起始和结束参数。我将“起始”设置为“5分钟前”，如下所示

from(bucket: "PlantBuddy")
  |> range(start: -5m)

现在，如果点击“提交”，我可以看到返回的最后 5 分钟的数据。

为了更方便地查看结果表格，我可以切换到“原始数据”视图。

现在我真正看到了返回的数据。

Flux 包含用于排序数据的函数，但我不特别需要这些。然而，我只想要土壤湿度读数。因此，我将这个表格传递到过滤函数中。我可以使用函数列表来获得一些帮助编写函数。它提供了一个占位符实现和有用的文档，直接在查询构建器中。

filter() 是一个接受另一个函数作为参数的函数。这种模式可能对各种 JavaScript 库的用户很熟悉。如果您熟悉匿名函数的创建，这将感觉一样。如果不熟悉，不要担心，从概念上和实践上都很简单。

以下是执行过滤的 Flux 代码行

|> filter(fn: (r) => r._measurement == "cpu")

我们可以稍微分解一下，以便更好地理解。 “filter” 是函数的名称。filter() 需要一个名为“fn”的命名参数，它是一个函数。这意味着我需要将另一个函数传递给 filter() 函数。我传递的函数将执行实际的过滤。 “r” 代表“记录”并是一个隐式变量，代表表格中的一行被传递到过滤实现中。实际的过滤只是一个无聊的布尔表达式。如果返回 true，则记录保留。否则，记录被丢弃。

当然，如果提交此查询，我不会得到任何结果，因为我的设备没有测量任何 CPU。我关心的是土壤湿度读数，这些读数在字段名为“soilMoisture”的行中被捕获。

我们可以看到它是如何工作的。因此，现在我可以添加另一个管道来计算所有这些字段的平均值。我可以搜索函数并得到一些帮助。

平均值函数默认使用 _value 字段，对我来说这很适用，所以我只需将其添加到我的代码中

from(bucket: "PlantBuddy")
|> range(start: -5m)
|> filter(fn: (r) => r._field == "soilMoisture")
|> mean()

现在，当我提交这个查询时，我在表格中得到一个单一值，这是迄今为止通过管道传递的所有行的平均值。

我还没有准备好将其移动到我的新桶中，因为 mean() 函数的结果删除了 _time 列，如果我在没有 _time 列的情况下尝试将结果添加到桶中，将会出现错误。mean() 函数确实添加了一个 _start 列和一个 _stop 列。我都不需要这些，所以我将简单地使用 rename() 函数将 _stop 列的名称更改为“_time”

from(bucket: "PlantBuddy")
  |> range(start: -5m)
  |> filter(fn: (r) => r._field == "soilMoisture")
  |> mean()
  |> rename(columns: {_stop: "_time"})

如果提交，我们可以看到现在有一个 _time 列。

现在，我只需要将其发送到我的桶中。由于您使用“from”将数据放入 Flux 脚本中，因此使用“to”导出数据是合乎逻辑的

from(bucket: "PlantBuddy")
  |> range(start: -5m)
  |> filter(fn: (r) => r._field == "soilMoisture")
  |> mean()
  |> rename(columns: {_stop: "_time"})
  |> to(bucket:"PlantBuddyCompressed", org:"[email protected]")

当我提交这个查询时，我得到“无结果”，因为我把所有数据都发送到了另一个桶。