使用时间序列数据，用开源工具助力你的边缘项目

Zoe Steinkamp 2023-03-28 原文

收集到的随时间变化的数据称为时间序列数据。今天，它已经成为每个行业和生态系统的一部分。它是不断增长的物联网行业的一大组成部分，将成为人们日常生活的重要部分。但时间序列数据及其需求很难处理。这是因为没有专门为处理时间序列数据而构建的工具。在这篇文章中，我将详细介绍这些问题，以及过去 10 年来 InfluxData 如何解决这些问题。

InfluxData

InfluxData 是一个开源的时间序列数据库平台。你可能通过 InfluxDB 了解该公司，但你可能不知道它专门从事时间序列数据库开发。这很重要，因为在管理时间序列数据时，你要处理两个问题：存储生命周期和查询。

在存储生命周期中，开发人员通常首先收集和分析非常详细的数据。但开发人员希望存储较小的、降低采样率的数据集，以描述其趋势，而不占用太多的存储空间。

查询数据库时，你不希望基于 ID 查询数据，而是希望基于时间范围进行查询。使用时间序列数据最常见的一件事是在一段时间内对其进行汇总。在典型的关系型数据库中存储数据时，这种查询是很慢的，这种数据库使用行和列来描述不同数据点的关系。专门为处理时间序列数据而设计的数据库可以更快地处理这类查询。InfluxDB 有自己的内置查询语言：Flux，这是专门为查询时间序列数据集而构建的。

Telegraf 如何工作的图像

数据采集

数据采集和数据处理都有一些很棒的工具。InfluxData 有 12 个以上的客户端库，允许你使用自己选择的编程语言编写和查询数据。这是自定义用法的一个很好的工具。开源摄取代理 Telegraf 包括 300 多个输入和输出插件。如果你是一个开发者，你也可以贡献自己的插件。

InfluxDB 还可以接受上传小体积历史数据集的 CSV 文件，以及大数据集的批量导入。

    import math
    bicycles3 = from(bucket: "smartcity")
        |> range(start:2021-03-01T00:00:00z, stop: 2021-04-01T00:00:00z)
        |> filter(fn: (r) => r._measurement == "city_IoT")
        |> filter(fn: (r) => r._field == "counter")
        |> filter(fn: (r) => r.source == "bicycle")
        |> filter(fn: (r) => r.neighborhood_id == "3")
        |> aggregateWindow(every: 1h, fn: mean, createEmpty:false)
    bicycles4 = from(bucket: "smartcity")
        |> range(start:2021-03-01T00:00:00z, stop: 2021-04-01T00:00:00z)
        |> filter(fn: (r) => r._measurement == "city_IoT")
        |> filter(fn: (r) => r._field == "counter")
        |> filter(fn: (r) => r.source == "bicycle")
        |> filter(fn: (r) => r.neighborhood_id == "4")
        |> aggregateWindow(every: 1h, fn: mean, createEmpty:false)join(tables: {neighborhood_3: bicycles3, neighborhood_4: bicycles4}, on ["_time"], method: "inner")
        |> keep(columns: ["_time", "_value_neighborhood_3","_value_neighborhood_4"])
        |> map(fn: (r) => ({
            r with
            difference_value : math.abs(x: (r._value_neighborhood_3 - r._value_neighborhood_4))
        }))

Flux

Flux 是我们的内部查询语言，从零开始建立，用于处理时间序列数据。它也是我们一些工具的基础动力，包括任务task、警报alert 和通知notification。要剖析上面的 Flux 查询，需要定义一些东西。首先，“桶bucket”就是我们所说的数据库。你可以配置存储桶，然后将数据流添加到其中。查询会调用 smartcity 存储桶，其范围为特定的一天（准确地说是 24 小时）。你可以从存储桶中获取所有数据，但大多数用户都包含一个数据范围。这是你能做的最基本的 Flux 查询。

接下来，我添加过滤器，将数据过滤到更精确、更易于管理的地方。例如，我过滤分配给 id 为 3 的社区中的自行车数量。从那里，我使用 aggregateWindow 获取每小时的平均值。这意味着我希望收到一个包含 24 列的表，每小时一列。我也对 id 为 4 的社区进行同样的查询。最后，我将这两张表相叠加，得出这两个社区自行车使用量的差异。

如果你想知道什么时候是交通高峰，这是不错的选择。显然，这只是 Flux 查询功能的一个小例子。但它提供了一个很好的例子，使用了 Flux 附带的一些工具。我还有很多的数据分析和统计功能。但对于这一点，我建议查看 Flux 文档。

    import "influxdata/influxdb/tasks"
    option task = {name: PB_downsample, every: 1h, offset: 10s}
    from(bucket: "plantbuddy")
        |>range(start: tasks.lastSuccess(orTime: -task.every))
        |>filter(fn: (r) => r["_measurement"] == "sensor_data")
        |>aggregateWindow(every: 10m, fn:last, createEmpty:false)
        |>yield(name: "last")
        |>to(bucket: "downsampled")

任务

InfluxDB 任务task 是一个定时 Flux 脚本，它接收输入数据流并以某种方式修改或分析它。然后，它将修改后的数据存储在新的存储桶中或执行其他操作。将较小的数据集存储到新的存储桶中，称为“降采样downsampling”，这是数据库的核心功能，也是时间序列数据生命周期的核心部分。

你可以在当前任务示例中看到，我已经对数据进行了降采样。我得到每 10 分钟增量的最后一个值，并将该值存储在降采样桶中。原始数据集在这 10 分钟内可能有数千个数据点，但现在降采样桶只有 60 个新值。需要注意的一点是，我还使用了范围内的 lastSuccess 函数。这会告诉 InfluxDB 从上次成功运行的时间开始运行此任务，以防它在过去 2 小时内失败，在这种情况下，它可以追溯 3 个小时内的最后一次成功运行。这对于内置错误处理非常有用。

检查和警报通知系统的图像

检查和警报

InfluxDB 包含一个警报Alert 或检查Check 和通知notification

许多人选择设置通知。为此，你需要定义一个通知端点notification endpoint。例如，聊天应用程序可以进行 HTTP 调用以接收通知。然后你定义何时接收通知，例如，你可以每小时运行一次检查。你可以每 24 小时运行一次通知。你可以让通知响应值更改，例如，“WARN”更改为“CRITICAL”，或者当值为“CRITICAL”时，无论如何都从“OK”更改为“WARN”。这是一个高度可定制的系统。从这个系统创建的 Flux 代码也可以编辑。

新 Edge 功能的图像

边缘

最后，我想把所有的核心功能放在一起，包括最近发布的一个非常特别的新功能。“Edge to cloud” 是一个非常强大的工具，允许你运行开源 InfluxDB，并在出现连接问题时在本地存储数据。连接修复后，它会将数据流传输到 InfluxData 云平台。

这对于边缘设备和重要数据非常重要，因为任何数据丢失都是有害的。你定义一个要复制到云的存储桶，然后该存储桶有一个磁盘支持的队列来本地存储数据。然后定义云存储桶应该复制到的内容。在连接到云端之前，数据都存储在本地。

InfluxDB 和物联网边缘

假设你有一个项目，你想使用连接到植物上的物联网传感器监测家里植物的健康状况。该项目是使用你的笔记本电脑作为边缘设备设置的。当你的笔记本电脑合上或关闭时，它会在本地存储数据，然后在重新连接时将数据流传到我的云存储桶。

图片展示了 Plant buddy 的工作方式

需要注意的一点是，在将数据存储到复制桶之前，这会对本地设备上的数据进行降采样。你的植物传感器每秒提供一个数据点。但它将数据压缩为一分钟的平均数，因此存储的数据更少了。在云账户中，你可以添加一些警报和通知，让你知道植物的水分何时低于某个水平，需要浇水。也可以在网站上使用视觉效果来告诉用户植物的健康状况。

数据库是许多应用程序的主干。在像 InfluxDB 的时间序列数据库平台中使用带有时间戳的数据可以节省开发人员的时间，并使他们能够访问各种工具和服务。InfluxDB 的维护者喜欢看到人们在我们的开源社区中构建什么，所以请与我们联系，并与其他人共享你的项目和代码！

助力使用时间 span style color 开源 $InfluxData

有关使用时间序列数据，用开源工具助力你的边缘项目的更多相关文章

ruby - 如何使用 Nokogiri 的 xpath 和 at_xpath 方法 - 2
我正在学习如何使用Nokogiri，根据这段代码我遇到了一些问题:require'rubygems'require'mechanize'post_agent=WWW::Mechanize.newpost_page=post_agent.get('http://www.vbulletin.org/forum/showthread.php?t=230708')puts"\nabsolutepathwithtbodygivesnil"putspost_page.parser.xpath('/html/body/div/div/div/div/div/table/tbody/tr/td/div
ruby - 使用 RubyZip 生成 ZIP 文件时设置压缩级别 - 2
我有一个Ruby程序，它使用rubyzip压缩XML文件的目录树。gem。我的问题是文件开始变得很重，我想提高压缩级别，因为压缩时间不是问题。我在rubyzipdocumentation中找不到一种为创建的ZIP文件指定压缩级别的方法。有人知道如何更改此设置吗？是否有另一个允许指定压缩级别的Ruby库？最佳答案这是我通过查看rubyzip内部创建的代码。level=Zlib::BEST_COMPRESSIONZip::ZipOutputStream.open(zip_file)do|zip|Dir.glob("**/*")d
ruby - 为什么我可以在 Ruby 中使用 Object#send 访问私有(private)/ protected 方法？ - 2
类classAprivatedeffooputs:fooendpublicdefbarputs:barendprivatedefzimputs:zimendprotecteddefdibputs:dibendendA的实例a=A.new测试a.foorescueputs:faila.barrescueputs:faila.zimrescueputs:faila.dibrescueputs:faila.gazrescueputs:fail测试输出failbarfailfailfail.发送测试[:foo,:bar,:zim,:dib,:gaz].each{|m|a.send(m)resc
ruby-on-rails - 使用 Ruby on Rails 进行自动化测试 - 最佳实践 - 2
很好奇，就使用rubyonrails自动化单元测试而言，你们正在做什么？您是否创建了一个脚本来在cron中运行rake作业并将结果邮寄给您？git中的预提交Hook？只是手动调用？我完全理解测试，但想知道在错误发生之前捕获错误的最佳实践是什么。让我们理所当然地认为测试本身是完美无缺的，并且可以正常工作。下一步是什么以确保他们在正确的时间将可能有害的结果传达给您？最佳答案不确定您到底想听什么，但是有几个级别的自动代码库控制:在处理某项功能时，您可以使用类似autotest的内容获得关于哪些有效，哪些无效的即时反馈。要确保您的提
ruby - 在 Ruby 中使用匿名模块 - 2
假设我做了一个模块如下:m=Module.newdoclassCendend三个问题:除了对m的引用之外，还有什么方法可以访问C和m中的其他内容？我可以在创建匿名模块后为其命名吗(就像我输入“module...”一样)？如何在使用完匿名模块后将其删除，使其定义的常量不再存在？最佳答案三个答案:是的，使用ObjectSpace.此代码使c引用你的类(class)C不引用m:c=nilObjectSpace.each_object{|obj|c=objif(Class===objandobj.name=~/::C$/)}当然这取决于
ruby - 使用 ruby 和 savon 的 SOAP 服务 - 2
我正在尝试使用ruby和Savon来使用网络服务。测试服务为http://www.webservicex.net/WS/WSDetails.aspx?WSID=9&CATID=2require'rubygems'require'savon'client=Savon::Client.new"http://www.webservicex.net/stockquote.asmx?WSDL"client.get_quotedo|soap|soap.body={:symbol=>"AAPL"}end返回SOAP异常。检查soap信封，在我看来soap请求没有正确的命名空间。任何人都可以建议我
python - 如何使用 Ruby 或 Python 创建一系列高音调和低音调的蜂鸣声？ - 2
关闭。这个问题是opinion-based.它目前不接受答案。想要改进这个问题？更新问题，以便editingthispost可以用事实和引用来回答它.关闭4年前。Improvethisquestion我想在固定时间创建一系列低音和高音调的哔哔声。例如:在150毫秒时发出高音调的蜂鸣声在151毫秒时发出低音调的蜂鸣声200毫秒时发出低音调的蜂鸣声250毫秒的高音调蜂鸣声有没有办法在Ruby或Python中做到这一点？我真的不在乎输出编码是什么(.wav、.mp3、.ogg等等)，但我确实想创建一个输出文件。
ruby-on-rails - 'compass watch' 是如何工作的/它是如何与 rails 一起使用的 - 2
我在我的项目目录中完成了compasscreate.和compassinitrails。几个问题:我已将我的.sass文件放在public/stylesheets中。这是放置它们的正确位置吗？当我运行compasswatch时，它不会自动编译这些.sass文件。我必须手动指定文件:compasswatchpublic/stylesheets/myfile.sass等。如何让它自动运行？文件ie.css、print.css和screen.css已放在stylesheets/compiled。如何在编译后不让它们重新出现的情况下删除它们？我自己编译的.sass文件编译成compiled/t
ruby - 使用 ruby 将 HTML 转换为纯文本并维护结构/格式 - 2
我想将html转换为纯文本。不过，我不想只删除标签，我想智能地保留尽可能多的格式。为插入换行符标签，检测段落并格式化它们等。输入非常简单，通常是格式良好的html(不是整个文档，只是一堆内容，通常没有anchor或图像)。我可以将几个正则表达式放在一起，让我达到80%，但我认为可能有一些现有的解决方案更智能。最佳答案首先，不要尝试为此使用正则表达式。很有可能你会想出一个脆弱/脆弱的解决方案，它会随着HTML的变化而崩溃，或者很难管理和维护。您可以使用Nokogiri快速解析HTML并提取文本:require'nokogiri'h
ruby - 在 64 位 Snow Leopard 上使用 rvm、postgres 9.0、ruby 1.9.2-p136 安装 pg gem 时出现问题 - 2
我想为Heroku构建一个Rails3应用程序。他们使用Postgres作为他们的数据库，所以我通过MacPorts安装了postgres9.0。现在我需要一个postgresgem并且共识是出于性能原因你想要pggem。但是我对我得到的错误感到非常困惑当我尝试在rvm下通过geminstall安装pg时。我已经非常明确地指定了所有postgres目录的位置可以找到但仍然无法完成安装:$envARCHFLAGS='-archx86_64'geminstallpg--\--with-pg-config=/opt/local/var/db/postgresql90/defaultdb/po