使用MASA Stack+.Net 从零开始搭建IoT平台 第五章 使用时序库存储上行数据

这篇具有很好参考价值的文章主要介绍了使用MASA Stack+.Net 从零开始搭建IoT平台 第五章 使用时序库存储上行数据。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

@

目录
  • 前言
  • 分析
  • 实施步骤
    • 时序库的安装
    • 解决playload没有时间戳问题
    • 代码编写
  • 测试
  • 总结

前言

我们可以将设备上行数据存储到关系型数据库中,我们需要两张带有时间戳的表(最新数据表历史数据表),历史数据表存储所有设备上报的数据,最新数据表需要存储设备最新一条上报数据,这条最新数据相当于设备的当前状态。然后展示的时候只展示最新一条数据的状态,报表查询可以按照设备id和时间从历史数据表查询汇总。
这样是可以的,但是我们的最新数据表需要被频繁的更新,数据量少的时候没问题。但数据量大,并发高的时候就会出现问题。
1、存储成本:数据不会被压缩,导致占用存储资源。
2、维护成本:单表数据量太大时,需要人工分库分表。
3、写入性能:单机写入吞吐量难以满足大量上行数据的写入需求,数据库存在性能瓶颈。
4、查询性能:数据量太大导致查询性能受到影响。

分析

我们可以采用时序库来解决上述问题,首先来了解一下什么是时序数据。时序数据是按照时间维度进行索引的数据,它记录了某个被测量实体在一定时间范围内,每个时间点上的一组测试值。传感器上传的室内PM2.5和甲醛数据、净水器传感器当前的TDS值、计算机系统的监控数据等,都属于时序数据,时序数据有如下特点:
1、数据量较大,写入操作是持续且平稳的,而且写多读少。
2、只有写入操作,几乎没有更新操作,比如去修改传感器的历史数据,是没有意义的。
3、没有随机删除,即使删除也是按照时间范围进行删除。删除某一个时间点的数据没有意义,但是删除2年前的数据是有意义的。
4、数据实时性和时效性强,数据随着时间的推移不断追加,旧数据很快失去意义。
5、大部分以时间和实体为维度进行查询,很少以测试值为维度查询,比如用户会查询某个时间段的温度数据,但是很少会去查询温度高于多少度的数据记录。
显然IoT的业务是符合使用时序库的场景的。
序数据库就是用来存储时序数据的数据库,时序数据库相较于传统的关系型数据和非关系型数据库而言,专门优化了对时序数据的存储,开源的时序数据库有InfluxDB OpenTSDB、TimeScaleDB 等。本文以InfluxDB数据库进行演示。
时序数据库有如下几个概念。
1.Metric:度量,相当于关系型数据库中的表(table)。
2.Data Point:数据点,相当于关系型数据库的中的行(row)。
3.Timestamp:时间戳,数据点生成时的时间戳。
4.Field:测量值,比如温度和湿度、PM2.5等。
5.Tag:标签,用于标识数据点,通常用来标识数据点的来源,比如温度和湿度数据来自哪个房间,哪个设备,可以当作关系型数据库表的主键。

如下图,度量为 Wind,每一个数据点都具有一个 timestamp,两个 field:direction 和 speed,两个 tag:sensor、city。它的第一行和第三行,存放的都是 sensor 号码为 95D8-7913 的设备,属性城市是上海。随着时间的变化,风向和风速都发生了改变,风向从 23.4 变成 23.2;而风速从 3.4 变成了 3.3。
使用MASA Stack+.Net 从零开始搭建IoT平台 第五章 使用时序库存储上行数据
图片来自网络

实施步骤

时序库的安装

安装参考官方文档,为了方便,我这里采用docker安装

docker run --name influxdb -p 8086:8086 influxdb:2.7.0

https://docs.influxdata.com/influxdb/v2.7/install/

我们打开 服务器ip:8086 可以看到它自带的管理界面,我们首先创建用户名密码,组织、以及Bucket的名称。
这里的bucket "IoTDemos" 相当于数据库的名称
使用MASA Stack+.Net 从零开始搭建IoT平台 第五章 使用时序库存储上行数据
我们记录一下这个Token,一会连接influxdb需要,相当于账号密码
使用MASA Stack+.Net 从零开始搭建IoT平台 第五章 使用时序库存储上行数据

解决playload没有时间戳问题

对于时序库来讲,时间戳是非常重要的,但是我们拿到的playload并没有时间戳(MQTTNet包我没有找到拿时间戳的方法)。
所以我们需要在mqtt上想办法,让设备上报数据的时候,mqtt自动添加时间戳到playload中。
1、我们在数据集成->规则中新建一条规则名称为"Add_Ts"。SQL编写如下

SELECT
  *,
  now_timestamp('millisecond') as payload.Ts
FROM
  "topic/#"

topic/# 代表消息发布到"topic/#"主题的事件
now_timestamp函数返回当前时间的 Unix 时间戳,我们将时间戳写入到payload的Ts属性中,关于更多内置SQL函数,请参考官方文档

https://www.emqx.io/docs/zh/v5.0/data-integration/rule-sql-builtin-functions.html

2、我们打开下面的调试,模拟设备上报一条数据,可以看到这条规则帮我们加入了时间戳。
使用MASA Stack+.Net 从零开始搭建IoT平台 第五章 使用时序库存储上行数据

3、然后我们还需要处理添加了时间戳的处理结果,我们在右侧添加一个动作,选择消息重发布,将刚刚添加了时间戳的消息重发到一个新的Topic上,我们使用topic/dp,并在playload中添加${payload},这样我们就修改了playload中的信息,添加了我们需要的时间戳,当然,我们Hub订阅的消息也需要对应修改,添加/dp后缀。

4、首先我们先修改MASA.IoT.Hub的配置文件,Topic添加"/dp"后缀

  "MqttSetting": {
...
    "Topic": "$share/IotHub/topic/+/dp"
  },

5、CallbackAsync中,因为我们设备名称是从Topic截取的,也要对应修改一下。

    private async Task CallbackAsync(MqttApplicationMessageReceivedEventArgs e)
    {
        var deviceDataPointStr = System.Text.Encoding.Default.GetString(e.ApplicationMessage.PayloadSegment);

        Console.WriteLine(deviceDataPointStr);
        var pubSubOptions = new PubSubOptions
        {
            //修改一下获取设备名称的方式
            DeviceName = e.ApplicationMessage.Topic[6..^3],
            Msg = deviceDataPointStr,
            PubTime = new DateTimeOffset(DateTime.Now).ToUnixTimeMilliseconds(),
            TrackId = Guid.NewGuid()
        };                            
...
    }

代码编写

解决完时间戳的问题,我们就可以编写代码向InfluxDB中写入数据了,我们首先在Infrastructure文件夹下创建ITimeSeriesDbClient接口和TimeSeriesDbClient类,使用接口也方便我们日后更换其他的时序库。
这里使用了InfluxDB.Client包。
ITimeSeriesDbClient.cs

namespace MASA.IoT.Core.Infrastructure
{
    public interface ITimeSeriesDbClient
    {
        bool WriteMeasurement<T>(T measurement);
    }
}

TimeSeriesDbClient.cs

using InfluxDB.Client;
using InfluxDB.Client.Api.Domain;
using MASA.IoT.WebApi;
using Microsoft.Extensions.Options;

namespace MASA.IoT.Core.Infrastructure
{
    public class TimeSeriesDbClient : ITimeSeriesDbClient
    {
        private readonly InfluxDBClient _client;
        private readonly string _bucket;
        private readonly string _org;
        private readonly AppSettings _appSettings;
        
        public TimeSeriesDbClient(IOptions<AppSettings> settings)
        {
            _appSettings = settings.Value;
            _org = _appSettings.InfluxDBSetting.Org;
            _bucket = _appSettings.InfluxDBSetting.Bucket;
            _client = new InfluxDBClient(_appSettings.InfluxDBSetting.Url, _appSettings.InfluxDBSetting.Token);
        }

        public bool WriteMeasurement<T>(T measurement)
        {
            try
            {
                using var writeApi = _client.GetWriteApi();
                writeApi.WriteMeasurement<T>(measurement, WritePrecision.Ms, _bucket, _org);
                return true;
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.Message);
                return false;
            }
        }
    }
}

这里使用new InfluxDBClient(_appSettings.InfluxDBSetting.Url, _appSettings.InfluxDBSetting.Token)来构造InfluxDBClient。
Token就是我们创建Bucket过程中保存的Token
Url是我们InfluxDB的访问地址:http://127.0.0.1:8086
写入的方法WriteMeasurement中我们通过_client.GetWriteApi创建一个写入的api然后直接将我们要写入的泛型实体写入,第二个可选参数代表写入精度,这里我们使用WritePrecision.Ms
我们在DeviceHandler.cs中注入ITimeSeriesDbClient 并添加一个WriteMeasurementAsync方法,在方法中我们先根据设备名称获取产品,如果识别产品ID为10001(空净产品),
那么我们就写入数据到Measurement:AirPurifierDataPoint
Measurement相当于数据库的表。
MeasurementColumn特性都是InfluxDB.Client.Core提供的,可以用来标识TagTimestamp

using InfluxDB.Client.Core;
using Newtonsoft.Json;

namespace MASA.IoT.Core.Contract
{
    [Measurement("AirPurifierDataPoint")]
    public class AirPurifierDataPoint
    {
        /// <summary>
        /// 设备名称
        /// </summary>
        [Column("DeviceName", IsTag = true)] public string DeviceName { get; set; }

        /// <summary>
        /// 产品ID
        /// </summary>
        [Column("ProductId", IsTag = true)] public Guid ProductId { get; set; }

        /// <summary>
        /// Pm2.5
        /// </summary>
        [Column("PM_25")] public double? Pm_25 { get; set; }
        /// <summary>
        /// 温度
        /// </summary>
        [Column("Temperature")] public double? Temperature { get; set; }
        /// <summary>
        /// 湿度
        /// </summary>
        [Column("Humidity")] public double? Humidity { get; set; }
        /// <summary>
        /// 时间戳
        /// </summary>
        [JsonProperty(propertyName: "Ts")]
        [Column(IsTimestamp = true)] public long Timestamp { get; set; }
    }
}

    public class DeviceHandler : IDeviceHandler
    {
        private readonly MASAIoTContext _ioTDbContext;
        private readonly IMqttHandler _mqttHandler;
        private readonly ITimeSeriesDbClient _timeSeriesDbClient;

        public DeviceHandler(MASAIoTContext ioTDbContext, IMqttHandler mqttHandler, ITimeSeriesDbClient timeSeriesDbClient)
        {
            _ioTDbContext = ioTDbContext;
            _mqttHandler = mqttHandler;
            _timeSeriesDbClient = timeSeriesDbClient;
        }

        /// <summary>
        /// 写入数据
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="pubSubOptions"></param>
        /// <returns></returns>
        public async Task<bool> WriteMeasurementAsync<T>(PubSubOptions pubSubOptions)
        {
            var device = await _ioTDbContext.IoTDeviceInfo.Include(o => o.ProductInfo).AsNoTracking()
                .FirstOrDefaultAsync(o => o.DeviceName == pubSubOptions.DeviceName);

            if (device != null && device.ProductInfo.ProductCode == "10001")  //空气净化器产品
            {
                var airPurifierDataPoint = JsonConvert.DeserializeObject<AirPurifierDataPoint>(pubSubOptions.Msg);

                airPurifierDataPoint.ProductId = device.ProductInfoId;
  
                return _timeSeriesDbClient.WriteMeasurement<AirPurifierDataPoint>(airPurifierDataPoint);

            }
            return false;
        }

除了WriteMeasurement方法之外,还提供了很多其他方法,如WritePoint,和批量写入的方法,可自行测试。

测试

我们启动项目,通过MQTTX向"topic/284202304230001"上报一条数据

{
  "DeviceName":"284202304230001",
  "Pm_25":100,
  "Temperature":25,
  "Humidity":50
}

我们在influxDB的管理工具中使用Data Explorer,使用如下的flux query查询语句,即可查出5分钟之内的数据,注意,这里的时间是UTC时间

如果想显示北京时区方便调试,可以在后面添加|> timeShift(duration: 8h)

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

使用MASA Stack+.Net 从零开始搭建IoT平台 第五章 使用时序库存储上行数据
关于flux查询语法

https://docs.influxdata.com/flux/v0.x/

总结

本节我们简单介绍了开源时序数据库influxDB的安装。
我们借助InfluxDB.Client库完成设备从上报到时序库数据存储的全过程,下一节我们介绍从时序库查询数据。

完整代码在这里:https://github.com/sunday866/MASA.IoT-Training-Demos文章来源地址https://www.toymoban.com/news/detail-500791.html

到了这里,关于使用MASA Stack+.Net 从零开始搭建IoT平台 第五章 使用时序库存储上行数据的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 从零开始,搭建一个简单的UVM验证平台(四)

    UVM前置基础: 1.UVM基础-factory机制、phase机制 2.UVM基础-组件(driver、monitor、agent...) 3.UVM基础-TLM通信机制(一) 4.UVM基础-TLM通信机制(二) ...还在更新 从零搭建一个UVM验证平台: 从零开始,搭建一个简单的UVM验证平台(一) 从零开始,搭建一个简单的UVM验证平台(二) 从

    2023年04月14日
    浏览(69)
  • 开发那点事(十六)从零开始搭建一个NFT数字藏品平台

    写在前面的话 从6月初到七月研究了将近一个月NFT 区块链这方面的东西,从啥都不会到了解原理,总算是有点成果了,在这里分享给大家。 核心大纲 百度超级链开放网络(Solidity语言) 集成openzeppelin中的ERC721合约快速完成合约开发 Springboot 作为后台开发语言调用线上合约 通

    2024年01月16日
    浏览(53)
  • Re.从零开始--基于UbuntuServer 20.04-OpenStack平台搭建_

    前言: 本文档基于ubuntu-server20.04版本和OpenStack Victoria搭建openstack环境 部署最小化Ubuntu-openstack满足基本服务;本文档均采用手动环境搭建 ubuntu源指定为阿里源,故搭环境需连接外网; ens33 ens34 节点名称 Ubuntu-controller 192.168.100.10 192.168.200.10 controller Ubuntu-compute 192.168.100.20 192.

    2024年01月20日
    浏览(49)
  • 从零开始,搭建边缘计算服务器并配置 Docker:只需三步(ARM平台)

    由于RK3588性能过剩(8+64配置),打算用安卓看电视+剩余服务器工作。 双系统方案思路:安卓上搭建Linux环境,部署Dokcer和Caddy代理。(比安卓+termux +qemu+docker 省了qemu环境,效率略微提高) 注1:RK3588 内核较高,无需改内核安装docker。 注2:部分安卓环境内核可能较低,需要自

    2024年01月16日
    浏览(50)
  • 从零开始在 Linux 上搭建 Hadoop 平台:一个详细的教程和必备配置文件

    在信息时代,大数据处理是企业发展必不可少的一部分。Hadoop 是一种分布式计算平台,能够高效地处理大数据集。然而,搭建 Hadoop 集群可以相当复杂,需要正确配置和调整多个组件。本文将向您展示如何从零开始在 Linux 上搭建 Hadoop,以及如何配置 Hadoop 的必备文件。 选择

    2024年02月08日
    浏览(53)
  • 提供最全面最详细的ESP32从零开始搭建一个物联网平台教程(从最基本的配网和内建WEB服务器开始到自已搭建一个MQTT服务器)

    目录 教程大纲  硬件需求 教程说明 教程章节链接 ESP32搭建WEB服务器一(AP配网) ESP32搭建WEB服务器二(STA模式) ESP32搭建WEB服务器三(AP模式与STA模式共存) ESP32搭建WEB服务器四(最简单的WEB服务器) ESP32搭建WEB服务器五(内嵌HTML) ESP32搭建WEB服务器六(利用SPIFFS存放html,css,js等文件(读取

    2024年02月13日
    浏览(62)
  • 从零开始的力扣刷题记录-第五十八天

    题目描述: 给你一个 不包含 任何零的整数数组 nums ,找出自身与对应的负数都在数组中存在的最大正整数 k 。 返回正整数 k ,如果不存在这样的整数,返回 -1 。 题解: 哈希表存储负数,再遍历nums对每一个正数去哈希表中查找是否存在对应的负数。存在就更新返回值 代码

    2024年02月09日
    浏览(46)
  • 从零开始搭建AI网站(6):如何使用响应式编程

    响应式编程(Reactive Programming)是一种编程范式,旨在处理异步数据流和事件流。它通过使用观察者模式和函数式编程的概念,将数据流和事件流抽象为可观察的序列,然后通过操作这些序列来实现各种功能。 在响应式编程中,数据流和事件流被视为连续的时间序列,可以通

    2024年02月10日
    浏览(39)
  • 从零开始的力扣刷题记录-第五十一天

    题目描述: 给你一棵二叉搜索树的 root ,请你 按中序遍历 将其重新排列为一棵递增顺序搜索树,使树中最左边的节点成为树的根节点,并且每个节点没有左子节点,只有一个右子节点。 题解: 中序遍历存储节点后按顺序连接即可 代码(Go): 题目描述: 小扣在秋日市集发

    2024年02月08日
    浏览(42)
  • 【Flutter】macOS从零开始使用FVM搭建Flutter开发环境

    本文为个人记录macOS系统使用fvm从零开始搭建flutter开发环境到项目运行的过程,非教程性质,仅供参考,如有疑问或建议,欢迎大家在评论区留言 附上开发设备配置 以vscode为编码工具 下载地址:Download Visual Studio Code - Mac, Linux, Windows 下载后解压安装即可 Xcode 在 iOS 真机或者

    2024年02月09日
    浏览(48)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包