DolphinDB海量数据查询：分页与采样

摘要

本文深入讲解DolphinDB海量数据查询技术。从分页查询到数据采样，从结果缓存到查询优化，全面介绍处理海量数据的核心方法。通过丰富的代码示例，帮助读者掌握海量数据查询的核心技能。

一、海量数据查询挑战

1.1 海量数据查询问题

工业物联网场景下，数据量巨大，查询面临挑战：

1.2 解决方案

二、分页查询

2.1 LIMIT分页

//创建测试数据 t=table(1..10000asid,2024.01.01+0..9999asdate,rand(20.0..30.0,10000)astemperature,rand(40.0..60.0,10000)ashumidity)//基本分页：每页100条 select*fromt limit100//分页查询：第2页（偏移100条） select*fromt limit100offset100//分页查询：第3页 select*fromt limit100offset200

2.2 TOP分页

//TOP分页：返回前N条 select top100*fromt//TOP分页带排序 select top100*fromt order byiddesc

2.3 分页函数

//分页查询函数defpaginateQuery(table,pageSize,pageNum){offset=pageSize*(pageNum-1)returnselect*fromtable limit pageSize offset offset}//使用分页函数 t=table(1..10000asid,rand(100.0,10000)asvalue)//第1页 paginateQuery(t,100,1)//第5页 paginateQuery(t,100,5)//第10页 paginateQuery(t,100,10)

2.4 分布式表分页

//创建分布式表 db=database("dfs://page_db",VALUE,1..100)schema=table(1:0,`device_id`timestamp`temperature`humidity,[INT,TIMESTAMP,DOUBLE,DOUBLE])db.createPartitionedTable(schema,`sensor_data,`device_id)//插入数据 loadTable("dfs://page_db","sensor_data").append!(table(take(1..100,100000)asdevice_id,take(now(),100000)astimestamp,rand(20.0..30.0,100000)astemperature,rand(40.0..60.0,100000)ashumidity))//分布式表分页查询 t=loadTable("dfs://page_db","sensor_data")//第1页 select*fromt limit100//第2页（注意：分布式表offset可能较慢） select*fromt limit100offset100//优化：使用分区条件 select*fromt where device_id=1limit100

2.5 分页最佳实践

三、数据采样

3.1 随机采样

//创建测试数据 t=table(1..100000asid,rand(100.0,100000)asvalue)//随机采样：采样10%select*fromt sample0.1//随机采样：采样1000条 select*fromt sample1000

3.2 系统采样

//系统采样：每隔N条取一条 select*fromt whereid%100=0//系统采样：每隔1000条取一条 select*fromt whereid%1000=0

3.3 分层采样

//创建带分类的数据 t=table(take(1..10,100000)asgroup_id,rand(100.0,100000)asvalue)//分层采样：每组采样100条 select*from(select*,row_number()over(partition by group_id order by rand())asrnfromt)where rn<=100

3.4 时间采样

//创建时间序列数据 t=table(2024.01.01T00:00:00+0..99999*60000astimestamp,//每分钟 rand(20.0..30.0,100000)astemperature)//时间采样：每小时取一条 select*fromt where minute(timestamp)=0//时间采样：使用bar函数 select bar(timestamp,1h)ashour,first(temperature)assample_tempfromt group by bar(timestamp,1h)

3.5 采样函数

//采样函数封装defsampleData(table,sampleRate){returnselect*fromtable sample sampleRate}defsampleByGroup(table,groupCol,sampleSize){returnselect*from(select*,row_number()over(partition by groupCol order by rand())asrnfromtable)where rn<=sampleSize}//使用采样函数 t=table(take(1..10,10000)asgroup_id,rand(100.0,10000)asvalue)//随机采样10%sampleData(t,0.1)//分组采样：每组100条 sampleByGroup(t,`group_id,100)

四、结果缓存

4.1 内存缓存

//查询结果缓存到内存 t=loadTable("dfs://page_db","sensor_data")//执行查询并缓存 cachedResult=select device_id,avg(temperature)asavg_tempfromt where date(timestamp)=2024.01.15group by device_id//使用缓存结果 select*fromcachedResult where avg_temp>25

4.2 共享表缓存

//使用共享表缓存 share cachedResultasshared_cache//其他会话可以使用 select*fromshared_cache

4.3 缓存更新策略

//定时更新缓存defupdateCache(){t=loadTable("dfs://page_db","sensor_data")result=select device_id,avg(temperature)asavg_tempfromt where date(timestamp)=today()-1group by device_id//更新共享缓存 shared_cache=result}//每日更新 scheduleJob("update_cache","更新缓存",updateCache,01:00,2024.01.01,2030.12.31,'D')

五、异步查询

5.1 异步查询原理

5.2 异步查询实现

//异步查询函数defasyncQuery(queryFunc){//提交查询 jobId=submitJob("async_query","异步查询",queryFunc)//返回作业IDreturnjobId}//查询作业状态defcheckQueryStatus(jobId){returngetJobStatus(jobId)}//获取查询结果defgetQueryResult(jobId){returngetJobReturn(jobId)}//使用异步查询 t=loadTable("dfs://page_db","sensor_data")//提交异步查询 jobId=asyncQuery(def(){returnselect count(*)fromloadTable("dfs://page_db","sensor_data")})//检查状态 checkQueryStatus(jobId)//获取结果 getQueryResult(jobId)

六、查询优化技巧

6.1 减少返回列

//不推荐：返回所有列 select*fromt limit1000//推荐：只返回需要的列 selectid,temperaturefromt limit1000

6.2 使用分区条件

//不推荐：全表扫描 select*fromt limit1000//推荐：分区裁剪 select*fromt where device_idin1..10limit1000

6.3 避免深分页

//不推荐：深分页（offset大） select*fromt limit100offset100000//推荐：使用游标 select*fromt whereid>100000//使用上一页最后一条的idlimit100

6.4 分批处理

//分批处理大数据defbatchProcess(tableName,batchSize,processFunc){t=loadTable(tableName)total=execcount(*)fromtfor(iin0..(total \ batchSize)){batch=select*fromt limit batchSize offset i*batchSize processFunc(batch)}}//使用分批处理 batchProcess("dfs://page_db/sensor_data",10000,def(batch){//处理每批数据print("处理批次: "+string(batch.rows()))})

七、实战案例

7.1 设备数据分页查询API

//设备数据分页查询APIdefqueryDeviceData(deviceId,startDate,endDate,pageSize,pageNum){t=loadTable("dfs://page_db","sensor_data")offset=pageSize*(pageNum-1)//查询总数 total=execcount(*)fromt where device_id=deviceIdanddate(timestamp)between startDateandendDate//分页查询 data=select*fromt where device_id=deviceIdanddate(timestamp)between startDateandendDate order by timestamp limit pageSize offset offset//返回结果returndict(STRING,ANY,[["total",total],["pageSize",pageSize],["pageNum",pageNum],["totalPages",ceil(total/pageSize)],["data",data]])}//使用API result=queryDeviceData(1,2024.01.01,2024.01.31,100,1)print("总数: "+string(result["total"]))print("数据: ")print(result["data"])

7.2 数据采样分析

//数据采样分析defanalyzeSampleData(sampleRate){t=loadTable("dfs://page_db","sensor_data")//采样数据 sample=select*fromt sample sampleRate//统计分析 stats=select device_id,count(*)ascnt,avg(temperature)asavg_temp,std(temperature)asstd_tempfromsample group by device_idreturnstats}//使用采样分析 analyzeSampleData(0.01)//1%采样

八、总结

本文详细介绍了DolphinDB海量数据查询：

1. 分页查询：LIMIT、TOP、分页函数
2. 数据采样：随机采样、系统采样、分层采样
3. 结果缓存：内存缓存、共享表缓存
4. 异步查询：异步提交、状态检查、结果获取
5. 查询优化：减少列、分区条件、避免深分页
6. 实战应用：分页API、采样分析

思考题：

1. 如何优化深分页查询？
2. 采样查询适合什么场景？
3. 如何设计高效的分页API？

参考资料

• DolphinDB分页查询
• DolphinDB数据采样