子域名做微信开放平台网站应用,phpcms 专题网站模板,北京建网站的公司哪个比较好,网站建设师薪资DolphinDB 高性能分布式时序数据库#xff0c;具有分布式计算、事务支持、多模存储、以及流批一体等能力#xff0c;非常适合作为一款理想的轻量级大数据平台#xff0c;轻松搭建一站式的高性能实时数据仓库。
本教程将以案例与脚本的方式#xff0c;介绍如何通过 Dolphin…DolphinDB 高性能分布式时序数据库具有分布式计算、事务支持、多模存储、以及流批一体等能力非常适合作为一款理想的轻量级大数据平台轻松搭建一站式的高性能实时数据仓库。
本教程将以案例与脚本的方式介绍如何通过 DolphinDB 快速搭建实时数仓助力各个行业如能源电力、航空航天、车联网、石油化工、矿业、智能制造、贸易政务、金融等在复杂业务场景下快速实现海量数据的低延时复杂指标计算和分析。
本教程包括原理介绍和实践操作配套示例代码用户可以根据教程结合自身业务特点动手搭建一个轻量级高性能的实时数据仓库。
1. 引言
1.1 案例背景与需求
随着大数据时代的来临各行各业对数据处理的实时性和准确性要求越来越高。传统的离线数仓虽然能够在一定程度上满足企业的数据存储和离线分析需求但在处理大规模实时数据时往往显得力不从心。尤其是在对数据实时性要求非常高的物联网和金融的头部企业离线数仓的局限性更加明显。
以电力行业的发电厂为例每个发电厂都拥有大量的测点这些测点实时采集着电站的运行数据。如何结合海量的电站运行数据对实时数据进行精准复杂的计算和分析成为了发电厂面临的一大挑战。传统的实时数据库欠缺对海量数据的聚合分析与计算能力而传统大数据系统搭建的离线数仓由于处理速度慢、时延高、架构复杂难以满足更深层次的业务需求。
DolphinDB 作为一款轻量级一站式实时数仓解决方案凭借其高性能分布式计算框架、实时流数据处理能力、分布式多模态存储引擎以及内存计算技术成为了解决这一问题的理想选择。
本文将通过 DolphinDB 实现一个典型的发电侧需求场景。在发电侧 4万个测点秒采样的情况下实时获取每个测点在1分钟、5分钟、30分钟、小时、天、月、直至最近1年的各项测点指标最大值、最小值、平均值、中位数、95%分位数、5%分位数、变化量、变化率、开始值、结束值等并且实现毫秒级查询响应。这些指标对于电站的运行监控、故障预警、能效分析大数据展示等方面至关重要。
1.2 数据仓库的基本概念
数据仓库Data Warehouse简称 DW 或 DWH是一种用于存储、处理和分析大量数据的系统旨在支持特定业务场景下的决策制定过程。数据仓库也是一种技术架构能够汇集并融合来自多个数据源如MySQL、Oracle、MongoDB、HBase 等的异构数据如数据表、Json、CSV、Protobuf 等通过数据清洗、集成和转换将数据整合至统一的存储体系如 DolphinDBHadoop中从而支持业务的多维分析、数据挖掘以及精准决策。 1.1 传统数据仓库典型架构图 数据仓库的重要性在于它能够帮助企业实现数据的集中管理和高效利用根据用途和实时性区分可以分为离线数仓和实时数仓两种类型。
离线数据仓库通常采用 T -1 的方式实现即每天定时如凌晨通过作业任务将前一天的历史数据导入数据仓库再通过 OLAPOnline Analytical Processing 对海量历史数据批数据进行分析查询。
对于大部分企业来说业务上迫切需要 T 0 实现实时风控、实时效果分析、实时过程管控等功能。传统离线数仓无法满足实时性要求因此出现了兼顾实时性和分析性的新型数据仓库架构即实时数仓。
实时数仓在技术要求和实现难度上要远远超过传统的数据仓库。相比传统数仓来说实时数仓可以更高效的数据处理能力和实时准实时的数据更新频率。在低延时的性能要求下需要解决数据源异构性、数据质量控制、事务和强一致性、多模存储、高性能聚合分析等技术难题。并且如何让普通开发人员具备实时数仓的开发和运维能力并持续稳定的进行产品迭代也是非常大的考验。
1.3 传统的实时数仓典型架构
传统的实时数仓通常以 Hadoop 大数据框架为基础使用 Lambda 架构或 Kappa 架构。技术复杂开发周期长无论在开发人员成本、时间成本还是硬件投入成本等方面来看对企业都是极大的负担。
传统实时数仓典型的技术栈如下所示
采集Sqoop、Flume、Flink CDC、DataX、Kafka存储HBase、HDFS、Hive、MySQL、MongoDB数据加工和计算Hive、Spark、Flink、Storm、PrestoOLAP 分析及查询TSDB/HTAP、ES、Kylin、DorisDB
企业要落地应用传统的实时数仓将会面临学习成本高、资源消耗大、扩展性和实时性不足等诸多问题。
1.4 DolphinDB 实时数仓架构与性能
与复杂的传统实时数仓不同DolphinDB 可通过自身产品能力快速实现轻量级实时数仓。既可独立进行采集、存储、流计算、ETL、决策分析与计算、可视化展示。亦可以作为企业已部署的各类第三方应用如大数据平台、AI 中台、驾驶舱的有效补充为企业级应用系统、集团级数据中台提供实时数仓的技术支撑以实现更复杂的应用场景。 DolphinDB 实时数仓业务架构图 DolphinDB 在物联网和金融等各行各业均拥有丰富且成熟的数据仓库实践案例充分展现了其广泛的应用价值。
以某省海关电子口岸公司的实时数仓项目为例DolphinDB 构建的实时数仓充分发挥了 All In One 轻量级一站式的产品优势。支持多源异构数据的接入兼容标准 SQL 支持复杂的多表关联强大的 ETL 数据清洗能力极大缩短了数据处理链条减少运维和开发成本。其业务架构及技术特点如下图所示 某省电子口岸实时数仓项目业务架构图 以下是在三机高可用集群部署的情况下DolphinDB 可支持的实时数仓性能指标参考
测点数量支持1亿测点写入吞吐量1亿测点/秒ODS 支持存储的记录数 1万亿条客户端最大连接数5000并发查询QPS 5000多维度聚合查询毫秒级实时流计算特征值提取50万/秒单条记录、单进程的删改软删除、upsert同步耗时≈ 10ms高可用集群多副本数据高可用多控制节点元数据高可用客户端断线重连和故障切换客户端高可用弹性扩展不停机水平扩展加节点、不停机垂直扩展加磁盘卷支持灰度升级
2. DolphinDB 实时数仓实践
接下来我们将以水电站发电机组设备实时监控的真实需求为案例通过 DolphinDB 搭建轻量级实时数仓。该案例可应用于能源电力、工业物联网、车联网等行业中。
欢迎大家动手尝试一起来验证一下
2.1 DolphinDB 安装部署
1. 下载官网社区最新版建议2.00.11及以上版本。
传送门https://cdn.dolphindb.cn/downloads/DolphinDB_Win64_V2.00.11.3.zip
2. windows 解压路径不能有空格避免安装到 Program Files 路径下。
官网教程https://docs.dolphindb.cn/zh/tutorials/deploy_dolphindb_on_new_server.html
3. 本次测试使用企业版license 可申请免费试用。如使用免费社区版建议降低测试的数据量级。
获取方式https://dolphindb.cn/product#downloads
4. 安装及测试过程中有任何问题可后台私信咨询。
2.2 实时数仓指标需求
数据基本情况
测点个数40000
采样频率秒级
计算指标聚合值 2.3 实践方案规划
以 DolphinDB 流计算框架为基础搭建边缘端轻量级实时数仓。所有计算结果在数据写入的同时高效完成时延控制在毫秒级。 对于1分钟计算周期、5分钟计算周期的指标以原始实时数据为基表对于30分钟计算周期、1小时计算周期的指标以1分钟计算结果为基表对于24小时计算周期指标以5分钟计算结果作为基表对于1年计算周期指标以24小时计算结果作为基表。
每类指标的计算窗口及滑动步长如下表所示
计算周期窗口长度滑动步长备注1分钟1分钟1分钟每间隔1分钟对过去1分钟窗口内的值进行计算5分钟5分钟5分钟每间隔5分钟对过去5分钟窗口内的值进行计算30分钟30分钟30分钟每间隔30分钟对过去30分钟窗口内的值进行计算1小时1小时1小时每间隔1小时对过去1小时窗口内的值进行计算24小时24小时24小时每间隔24小时对过去24小时窗口内的值进行计算1年1年24小时每间隔1天对过去1年窗口内的值进行计算
3. 性能测试及结果
3.1 测试环境
为了方便测试和验证采用单机单节点的部署方式实现轻量级实时数仓服务器配置如下
CPU12核内存32GB磁盘1.1T HDD 150MB/s
通过脚本模拟全量测点4000024小时内的实时数据2023.01.01T00:00:00—2023.01.02T00:00:01.000进行 1分钟、5分钟、30分钟、1小时、24小时的窗口聚合计算并将计算结果写入分布式数据库。在某个窗口内数据条数有可能比窗口长度要小
对于1年窗口的计算另模拟24小时窗口计算结果实时数据对该模拟的结果进行实时聚合计算。
详细的测试脚本包含在文末的附件中。
3.2 测试结果
性能测试结果见下表 注上表中全部测点计算耗时为时间窗口内所有测点指标计算的耗时单测点/多测点计算耗时为时间窗口内所选测点指标计算的耗时。
4. 总结
通过本教程的学习和实践我们深入了解了 DolphinDB 在构建轻量级实时数仓方面的强大能力。DolphinDB 以其高性能、分布式、实时计算的特点为各个行业提供了快速实现海量数据低延时复杂指标计算和分析的有力工具。
通过实践操作我们可以体验到了 DolphinDB 的易用性和高效性。无论是数据导入、数据查询还是复杂的流式计算DolphinDB 都提供了简洁明了的语法和强大的功能。附件中所提供的脚本不仅包括 DolphinDB 的基本使用和操作方法更能够深入了解实时数仓的构建原理和应用场景。这使得我们可以快速构建出符合业务需求的实时数仓并实时响应各种复杂的分析需求。
最后希望读者能够结合本教程的示例代码和自身业务特点动手搭建一个轻量级高性能的实时数据仓库。在实际应用中不断挖掘 DolphinDB 的潜力无论是能源电力、石油化工、智能制造、航空航天还是车联网、金融等行业DolphinDB 均可为实时数仓的广泛应用提供有力支持。
5. 附件
测试结果可通过以下脚本在 DolphinDB 服务器上进行复现
def clearEnv(){//取消订阅unsubscribeTable(tableNameinputStream, actionNamedispatch1)unsubscribeTable(tableNameinputStream, actionNamedispatch2)unsubscribeTable(tableNameoneMinuteResult, actionNamecalcHalfHour)unsubscribeTable(tableNameoneMinuteResult, actionNamecalcOneHour)unsubscribeTable(tableNamefiveMinuteResult, actionNamecalcOneDay)unsubscribeTable(tableName oneDayResultSimulate,actionNamecalcOneYear)unsubscribeTable(tableName oneMinuteResult,actionNameappendInToDFS)unsubscribeTable(tableName fiveMinuteResult,actionNameappendInToDFS)unsubscribeTable(tableName halfHourResult,actionNameappendInToDFS)unsubscribeTable(tableName oneHourResult,actionNameappendInToDFS)unsubscribeTable(tableName oneDayResult,actionNameappendInToDFS)unsubscribeTable(tableName oneYearResult,actionNameappendInToDFS)//删除流计算引擎for(i in 1..2){try{dropStreamEngine(dispatchDemostring(i))}catch(ex){print(ex)}}for(i in 1..5){try{dropStreamEngine(oneMinuteCalcstring(i))}catch(ex){print(ex)}try{dropStreamEngine(fiveMinuteCalcstring(i))}catch(ex){print(ex)}}try{dropStreamEngine(halfHourCalc)}catch(ex){print(ex)}try{dropStreamEngine(oneHourCalc)}catch(ex){print(ex)}try{dropStreamEngine(oneDayCalc)}catch(ex){print(ex)}try{dropStreamEngine(oneYearCalc)}catch(ex){print(ex)}//删除流数据表try{dropStreamTable(inputStream)}catch(ex){print(ex)}try{dropStreamTable(oneMinuteResult)}catch(ex){print(ex)}try{dropStreamTable(fiveMinuteResult)}catch(ex){print(ex)}try{dropStreamTable(halfHourResult)}catch(ex){print(ex)}try{dropStreamTable(oneHourResult)}catch(ex){print(ex)}try{dropStreamTable(oneDayResult)}catch(ex){print(ex)}try{dropStreamTable(oneDayResultSimulate)}catch(ex){print(ex)}try{dropStreamTable(oneYearResult)}catch(ex){print(ex)}}def createStreamTable(){//定义输入流表enableTableShareAndPersistence(table streamTable(1000:0,TimedeviceIdvalue,TIMESTAMPSYMBOLDOUBLE),tableName inputStream,cacheSize 1000000,precache1000000)colName TimedeviceIdfilterTimeMAXMINMEANMEDP95P5CHANGECHANGE_RATEfirstlastendTimecolType TIMESTAMPSYMBOLNANOTIMESTAMP join take(DOUBLE,10) join NANOTIMESTAMP//定义1分钟窗口计算结果流表 enableTableShareAndPersistence(table streamTable(1000:0,colName,colType),tableName oneMinuteResult,cacheSize 1000000,precache1000000)//定义5分钟窗口计算结果流表 enableTableShareAndPersistence(table streamTable(1000:0,colName,colType),tableName fiveMinuteResult,cacheSize 1000000,precache1000000)//定义30分钟窗口计算结果流表 enableTableShareAndPersistence(table streamTable(1000:0,colName,colType),tableName halfHourResult,cacheSize 1000000,precache1000000)//定义1小时窗口计算结果流表 enableTableShareAndPersistence(table streamTable(1000:0,colName,colType),tableName oneHourResult,cacheSize 1000000,precache1000000)//定义24小时窗口计算结果流表 enableTableShareAndPersistence(table streamTable(1000:0,colName,colType),tableName oneDayResult,cacheSize 1000000,precache1000000)//定义模拟24小时窗口计算结果流表colName TIMEdeviceIdMAXMINMEANMEDP95P5CHANGECHANGE_RATEfirstlastcolType DATESYMBOL join take(DOUBLE,10) enableTableShareAndPersistence(table streamTable(1000:0,colName,colType),tableName oneDayResultSimulate,cacheSize 1000000,precache1000000)//定义1年窗口计算结果流表colName TimedeviceIdfilterTimeMAXMINMEANMEDP95P5CHANGECHANGE_RATEfirstlastendTimecolType DATESYMBOLNANOTIMESTAMP join take(DOUBLE,10) join NANOTIMESTAMP enableTableShareAndPersistence(table streamTable(1000:0,colName,colType),tableName oneYearResult,cacheSize 1000000,precache1000000)
}def createDFS(){//创建存储计算1分钟窗口计算结果表if(existsDatabase(dfs://oneMinuteCalc)){dropDatabase(dfs://oneMinuteCalc)}db1 database(, VALUE,2023.01.01..2023.01.03)db2 database(, HASH,[SYMBOL,20])db database(directorydfs://oneMinuteCalc, partitionTypeCOMPO, partitionScheme[db1,db2],engineTSDB)colName TimedeviceIdfilterTimeMAXMINMEANMEDP95P5CHANGECHANGE_RATEfirstlastendTimecolType TIMESTAMPSYMBOLNANOTIMESTAMP join take(DOUBLE,10) join NANOTIMESTAMPt table(1:0,colName,colType)pt db.createPartitionedTable(tablet,tableName test ,partitionColumns [Time,deviceId],sortColumns [deviceId,Time],compressMethods{Time:delta})//创建存储计算5分钟窗口计算结果表 if(existsDatabase(dfs://fiveMinuteCalc)){dropDatabase(dfs://fiveMinuteCalc)}db database(directorydfs://fiveMinuteCalc, partitionTypeVALUE,partitionScheme2023.01.01..2023.01.03,engineTSDB)t table(1:0,colName,colType)pt db.createPartitionedTable(tablet,tableName test ,partitionColumns [Time],sortColumns [deviceId,Time],compressMethods{Time:delta},sortKeyMappingFunction[hashBucket{,100}]) //创建存储计算30分钟窗口计算结果表if(existsDatabase(dfs://halfHourCalc)){dropDatabase(dfs://halfHourCalc)}db database(directorydfs://halfHourCalc, partitionTypeVALUE, partitionScheme2023.01.01..2023.01.03,engineTSDB)t table(1:0,colName,colType)pt db.createPartitionedTable(tablet,tableName test ,partitionColumns [Time],sortColumns [deviceId,Time],compressMethods{Time:delta}) //创建存储计算1小时窗口计算结果表if(existsDatabase(dfs://oneHourCalc)){dropDatabase(dfs://oneHourCalc)}db database(directorydfs://oneHourCalc, partitionTypeVALUE,partitionScheme2023.01.01..2023.01.03,engineTSDB)t table(1:0,colName,colType)pt db.createPartitionedTable(tablet,tableName test ,partitionColumns [Time],sortColumns [deviceId,Time],compressMethods{Time:delta})//创建存储计算24小时窗口计算结果表if(existsDatabase(dfs://oneDayCalc)){dropDatabase(dfs://oneDayCalc)}db database(directorydfs://oneDayCalc, partitionTypeVALUE, partitionScheme2023.01.01..2023.01.03,engineTSDB)t table(1:0,colName,colType)pt db.createPartitionedTable(tablet,tableName test ,partitionColumns [Time],sortColumns [deviceId,Time],compressMethods{Time:delta})//创建存储计算1年窗口计算结果表if(existsDatabase(dfs://oneYearCalc)){dropDatabase(dfs://oneYearCalc)}db database(directorydfs://oneYearCalc, partitionTypeVALUE, partitionScheme2023.01.01..2023.01.03,engineTSDB)t table(1:0,colName,colType)pt db.createPartitionedTable(tablet,tableName test ,partitionColumns [Time],sortColumns [deviceId,Time],compressMethods{Time:delta})
}//1分钟窗口计算过滤函数
def filter1(msg){t select *,now(true) as filterTime from msg getStreamEngine(dispatchDemo1).append!(t)
}//5分钟窗口计算过滤函数
def filter2(msg){t select *,now(true) as filterTime from msg getStreamEngine(dispatchDemo2).append!(t)
}//30分钟窗口计算过滤函数
def filter3(msg){t select *,now(true) as filterTime2 from msg getStreamEngine(halfHourCalc).append!(t)
}//1小时窗口计算
def filter4(msg){t select *,now(true) as filterTime2 from msg getStreamEngine(oneHourCalc).append!(t)
}//24小时窗口计算
def filter5(msg){t select *,now(true) as filterTime2 from msg getStreamEngine(oneDayCalc).append!(t)
}clearEnv();
createStreamTable();
createDFS();schemas1 table(1:0,TimedeviceIdvaluefilterTime,TIMESTAMPSYMBOLDOUBLENANOTIMESTAMP)
metrics1 [first(filterTime),max(value),min(value),mean(value),med(value),percentile(value,95),percentile(value,5),last(value)-first(value),(last(value)-first(value))/first(value),first(value),last(value),now(true)]
//创建1分钟窗口聚合计算引擎
for(i in 1..5){engine1 createTimeSeriesEngine(nameoneMinuteCalcstring(i), windowSize60000, step60000,metricsmetrics1 , dummyTableschemas1 , outputTableobjByName(oneMinuteResult),timeColumn Time, useSystemTimefalse, keyColumn deviceId)
}//创建5分钟窗口聚合计算引擎
for(i in 1..5){engine2 createTimeSeriesEngine(namefiveMinuteCalcstring(i), windowSize300000, step300000, metricsmetrics1 , dummyTableschemas1 , outputTableobjByName(fiveMinuteResult),timeColumn Time, useSystemTimefalse, keyColumn deviceId)
}//1分钟、5分钟窗口聚合计算分发引擎
dispatchEngine1createStreamDispatchEngine(namedispatchDemo1, dummyTableschemas1, keyColumndeviceId, outputTable[getStreamEngine(oneMinuteCalc1),getStreamEngine(oneMinuteCalc2),getStreamEngine(oneMinuteCalc3),getStreamEngine(oneMinuteCalc4),getStreamEngine(oneMinuteCalc5)])
dispatchEngine2createStreamDispatchEngine(namedispatchDemo2, dummyTableschemas1, keyColumndeviceId, outputTable[getStreamEngine(fiveMinuteCalc1),getStreamEngine(fiveMinuteCalc2),getStreamEngine(fiveMinuteCalc3),getStreamEngine(fiveMinuteCalc4),getStreamEngine(fiveMinuteCalc5)])colName TimedeviceIdfilterTimeMAXMINMEANMEDP95P5CHANGECHANGE_RATEfirstlastendTimefilterTime2
colType TIMESTAMPSYMBOLNANOTIMESTAMP join take(DOUBLE,10) join NANOTIMESTAMPNANOTIMESTAMP
schemas2 table(1:0,colName,colType)
metrics2 [first(filterTime2),max(MAX),min(MIN),mean(MEAN),med(MED),avg(P95),avg(P5),last(last)-first(first),(last(last)-first(first))/first(first),first(first),last(last),now(true)]
//创建30分钟窗口聚合计算引擎
engine3 createTimeSeriesEngine(namehalfHourCalc, windowSize1800000, step1800000, metricsmetrics2 , dummyTableschemas2 , outputTableobjByName(halfHourResult),timeColumn Time, useSystemTimefalse, keyColumn deviceId)//创建1小时窗口聚合计算引擎
engine4 createTimeSeriesEngine(nameoneHourCalc, windowSize3600000, step3600000, metricsmetrics2 , dummyTableschemas2 , outputTableobjByName(oneHourResult),timeColumn Time, useSystemTimefalse, keyColumn deviceId)//创建24小时窗口聚合计算引擎
engine5 createTimeSeriesEngine(nameoneDayCalc, windowSize86400000, step86400000, metricsmetrics2 , dummyTableschemas2 , outputTableobjByName(oneDayResult),timeColumn Time, useSystemTimefalse, keyColumn deviceId)//订阅
subscribeTable(tableNameinputStream, actionNamedispatch1, handlerfilter1, msgAsTable true,batchSize 10240)
subscribeTable(tableNameinputStream, actionNamedispatch2, handlerfilter2, msgAsTable true,batchSize 10240)subscribeTable(tableNameoneMinuteResult, actionNamecalcHalfHour, handlerfilter3,msgAsTable true,batchSize 10240)
subscribeTable(tableNameoneMinuteResult, actionNamecalcOneHour, handlerfilter4, msgAsTable true,batchSize 10240)
subscribeTable(tableNamefiveMinuteResult, actionNamecalcOneDay, handlerfilter5, msgAsTable true,batchSize 10240)subscribeTable(tableName oneMinuteResult,actionNameappendInToDFS,offset0,handlerloadTable(dfs://oneMinuteCalc,test),msgAsTabletrue,batchSize10240)
subscribeTable(tableName fiveMinuteResult,actionNameappendInToDFS,offset0,handlerloadTable(dfs://fiveMinuteCalc,test),msgAsTabletrue,batchSize10240)
subscribeTable(tableName halfHourResult,actionNameappendInToDFS,offset0,handlerloadTable(dfs://halfHourCalc,test),msgAsTabletrue,batchSize10240)
subscribeTable(tableName oneHourResult,actionNameappendInToDFS,offset0,handlerloadTable(dfs://oneHourCalc,test),msgAsTabletrue,batchSize10240)
subscribeTable(tableName oneDayResult,actionNameappendInToDFS,offset0,handlerloadTable(dfs://oneDayCalc,test),msgAsTabletrue,batchSize10240)def filter6(msg){tmp select * ,now(true) as filterTime from msg getStreamEngine(oneYearCalc).append!(tmp)
} colName TimedeviceIdMAXMINMEANMEDP95P5CHANGECHANGE_RATEfirstlastfilterTime
colType DATESYMBOL join take(DOUBLE,10) join NANOTIMESTAMP
schemas3 table(1:0,colName,colType)
metrics3 [last(filterTime),max(MAX),min(MIN),mean(MEAN),med(MED),avg(P95),avg(P5),last(last)-first(first),(last(last)-first(first))/first(first),first(first),last(last),now(true)]
engine6 createTimeSeriesEngine(nameoneYearCalc, windowSize365, step1, metricsmetrics3 , dummyTableschemas3 , outputTableobjByName(oneYearResult),timeColumn Time, useSystemTimefalse, keyColumn deviceId)subscribeTable(tableName oneDayResultSimulate,actionNamecalcOneYear, handlerfilter6, msgAsTable true,batchSize 10240)
subscribeTable(tableName oneYearResult,actionNameappendInToDFS,offset0,handlerloadTable(dfs://oneYearCalc,test),msgAsTabletrue)deviceIdList lapd(string(rand(10000,700)),6,0) //测点id//模拟数据的函数一共模拟1小时的数据
def simulateData(deviceIdList){num deviceIdList.size()startTime timestamp(2023.01.01)do{Time take(startTime,num)deviceId deviceIdListvalue rand(100.0,num)objByName(inputStream).append!(table(Time,deviceId,value))startTime startTime1000sleep(100)}while(startTime2023.01.02T00:00:10.000)
}def simulateOneDay(deviceIdList){num deviceIdList.size()startTime 2022.01.01do{Time take(startTime,num)deviceId deviceIdListMAX rand(100.0,num)MIN rand(100.0,num)MEAN rand(100.0,num)MED rand(100.0,num)P95 rand(100.0,num)P5 rand(100.0,num)CHANGE rand(100.0,num)CHANGE_RATE rand(100.0,num)first rand(100.0,num)last rand(100.0,num)tmp table(Time,deviceId,MAX,MIN,MEAN,MED,P95,P5,CHANGE,CHANGE_RATE,first,last)objByName(oneDayResultSimulate).append!(tmp)startTime startTime1sleep(500)}while(startTime2023.12.31)
}submitJob(simulateData,write,simulateData,deviceIdList)
submitJob(simulateOneDay,write,simulateOneDay,deviceIdList)//耗时统计
tmp1 select Time,deviceId,filterTime,endTime from loadTable(dfs://oneYearCalc,test) order by Time,deviceId
tmp2 select Time,deviceId,next(filterTime) as startTime,endTime from tmp1 context by deviceId
select avg(endTime-startTime)\1000\1000 as timeUsed from tmp2 group by deviceId //统计单个测点的计算耗时
tmp3 select min(startTime) as st,max(endTime) as dt from tmp2 group by Time
select (dt-st)\1000\1000 as used from tmp3 //统计整个时间窗口的计算耗时
