DolphinDB分布式表：创建与管理

目录摘要一、分布式表概述1.1 什么是分布式表1.2 分布式表特点1.3 分布式表 vs 内存表二、创建分布式表2.1 创建分布式数据库2.2 创建分布式表2.3 创建维度表三、数据写入3.1 批量写入3.2 并行写入3.3 流式写入四、数据查询4.1 基本查询4.2 分区裁剪4.3 聚合查询4.4 分布式计算五、表管理5.1 查看表信息5.2 更新数据5.3 删除数据5.4 删除表六、分区管理6.1 查看分区6.2 添加分区6.3 删除分区七、副本管理7.1 副本配置7.2 副本状态八、实战案例8.1 工业物联网数据平台九、总结参考资料摘要本文深入讲解DolphinDB分布式表的创建与管理。从分布式表原理到创建方法从数据写入到查询优化从分区管理到运维监控全面介绍分布式表的核心操作。通过丰富的代码示例帮助读者掌握分布式表的设计和管理技能。一、分布式表概述1.1 什么是分布式表分布式表是存储在分布式数据库中的表数据按分区分布在多个节点上分布式表架构客户端ControllerDataNode1DataNode2DataNode3分区1,3,5分区2,4,6分区7,9,111.2 分布式表特点特点说明分布式存储数据分布在多个节点水平扩展支持PB级数据高可用多副本机制并行计算分布式计算加速1.3 分布式表 vs 内存表特性内存表分布式表存储位置内存磁盘容量受内存限制PB级持久化需手动自动持久化查询速度快较快适用场景临时数据海量数据二、创建分布式表2.1 创建分布式数据库//创建VALUE分区数据库 dbdatabase(dfs://iot_db,VALUE,1..100)//创建RANGE分区数据库 dbdatabase(dfs://iot_db,RANGE,2024.01.01..2024.12.31)//创建COMPO组合分区数据库 dbdatabase(dfs://iot_db,COMPO,[RANGE,2024.01.01..2024.12.31,HASH,[INT,10]])2.2 创建分布式表//创建分布式数据库 dbdatabase(dfs://iot_db,VALUE,1..100)//定义表结构 schematable(1:0,device_idtimestamptemperaturehumiditypressure,[INT,TIMESTAMP,DOUBLE,DOUBLE,DOUBLE])//创建分布式表 db.createPartitionedTable(schema,sensor_data,device_id)//查看表结构 schema(loadTable(dfs://iot_db,sensor_data))2.3 创建维度表//维度表小表广播到所有节点 dbdatabase(dfs://metadata)//创建设备信息维度表 schematable(1:0,device_iddevice_namelocationinstall_datestatus,[INT,STRING,STRING,DATE,SYMBOL])db.createTable(schema,device_info)//插入数据 loadTable(dfs://metadata,device_info).append!(table(1..100asdevice_id,device_string(1..100)asdevice_name,take(车间A车间B车间C,100)aslocation,take(2024.01.01,100)asinstall_date,take(在线,100)asstatus))三、数据写入3.1 批量写入//批量写入数据 ttable(take(1..100,10000)asdevice_id,take(now(),10000)astimestamp,rand(20.0..30.0,10000)astemperature,rand(40.0..60.0,10000)ashumidity,rand(1000.0..1020.0,10000)aspressure)loadTable(dfs://iot_db,sensor_data).append!(t)3.2 并行写入//并行写入函数defparallelWrite(dbPath,tableName,data){loadTable(dbPath,tableName).append!(data)}//使用mr并行写入 data_listsplit(t,10)//分成10份 mr(data_list,parallelWrite{dfs://iot_db,sensor_data})3.3 流式写入//创建流表 share streamTable(1:0,device_idtimestamptemperaturehumiditypressure,[INT,TIMESTAMP,DOUBLE,DOUBLE,DOUBLE])asinput_stream//订阅流表写入分布式表 subscribeTable(,input_stream,persist,-1,def(msg){loadTable(dfs://iot_db,sensor_data).append!(msg)},true,10000,true)//写入流表 input_stream.append!(t)四、数据查询4.1 基本查询//加载分布式表 tloadTable(dfs://iot_db,sensor_data)//查询数据 select count(*)fromt select top100*fromt select*fromt where device_id14.2 分区裁剪//分区裁剪优化查询 tloadTable(dfs://iot_db,sensor_data)//按分区列过滤高效 select count(*)fromt where device_idin[1,2,3]//查看执行计划 explain select*fromt where device_id14.3 聚合查询//分组聚合 select device_id,count(*)ascnt,avg(temperature)asavg_temp,max(temperature)asmax_temp,min(temperature)asmin_tempfromt group by device_id//时间窗口聚合 select bar(timestamp,1h)ashour,avg(temperature)asavg_tempfromt where device_id1group by bar(timestamp,1h)4.4 分布式计算//分布式计算使用mr函数defmapFunc(data){returnselect device_id,avg(temperature)asavg_tempfromdata group by device_id}defreduceFunc(results){returnunionAll(results)}//执行分布式计算 mr(sqlDS(select*fromt),mapFunc,reduceFunc)五、表管理5.1 查看表信息//查看表结构 schema(loadTable(dfs://iot_db,sensor_data))//查看表大小 getTabletsMeta(dfs://iot_db,sensor_data)//查看分区分布 select partition,count(*)ascntfromgetTabletsMeta(dfs://iot_db,sensor_data)group by partition5.2 更新数据//更新分布式表 tloadTable(dfs://iot_db,sensor_data)//条件更新 update tsettemperaturetemperature*1.1where device_id1//注意更新操作需要指定分区条件5.3 删除数据//删除数据 tloadTable(dfs://iot_db,sensor_data)//条件删除 deletefromt where device_id1//清空表 truncate(t)5.4 删除表//删除分布式表 dropTable(database(dfs://iot_db),sensor_data)//删除数据库 dropDatabase(dfs://iot_db)六、分区管理6.1 查看分区//查看分区方案 dbdatabase(dfs://iot_db)db.partitionSchema()//查看分区详情 getTabletsMeta(dfs://iot_db,sensor_data)6.2 添加分区//添加新分区RANGE分区 dbdatabase(dfs://iot_db)addPartitions(db,[2025.01.01,2025.02.01])6.3 删除分区//删除分区 dbdatabase(dfs://iot_db)dropPartition(db,[2023.01.01])七、副本管理7.1 副本配置//创建带副本的数据库 dbdatabase(dfs://iot_db_replica,VALUE,1..100,replica2)//查看副本配置 db.replicationFactor()7.2 副本状态//查看副本状态 getClusterReplicationStatus()//查看副本同步状态 getReplicationDelay()八、实战案例8.1 工业物联网数据平台//1.创建分布式数据库dbdatabase(dfs://iot_platform,COMPO,[RANGE,2024.01.01..2024.12.31,HASH,[INT,100]],replica2)//2.创建设备数据表schematable(1:0,device_idtimestamptemperaturehumiditypressurevibrationpowerstatus,[INT,TIMESTAMP,DOUBLE,DOUBLE,DOUBLE,DOUBLE,DOUBLE,SYMBOL])db.createPartitionedTable(schema,device_data,timestampdevice_id)//3.创建告警表schematable(1:0,alert_iddevice_idtimestampalert_typealert_levelmessage,[LONG,INT,TIMESTAMP,SYMBOL,INT,STRING])db.createPartitionedTable(schema,alert_data,timestampdevice_id)//4.创建统计表schematable(1:0,device_iddateavg_tempmax_tempmin_temprecord_count,[INT,DATE,DOUBLE,DOUBLE,DOUBLE,LONG])db.createPartitionedTable(schema,daily_stats,datedevice_id)//5.数据写入函数defwriteDeviceData(data){loadTable(dfs://iot_platform,device_data).append!(data)}//6.统计计算函数defcalcDailyStats(){tloadTable(dfs://iot_platform,device_data)statsselect device_id,date(timestamp)asdate,avg(temperature)asavg_temp,max(temperature)asmax_temp,min(temperature)asmin_temp,count(*)asrecord_countfromt where date(timestamp)today()-1group by device_id,date(timestamp)loadTable(dfs://iot_platform,daily_stats).append!(stats)}//每日统计 scheduleJob(daily_stats,每日统计,calcDailyStats,00:30,2024.01.01,2030.12.31,D)print(工业物联网数据平台创建完成)九、总结本文详细介绍了DolphinDB分布式表的创建与管理分布式表原理分布式存储、水平扩展、高可用创建方法创建数据库、创建表、维度表数据写入批量写入、并行写入、流式写入数据查询基本查询、分区裁剪、分布式计算表管理查看信息、更新删除、分区管理副本管理副本配置、副本状态思考题分布式表和内存表有什么区别如何设计高效的分区策略如何优化分布式表查询性能参考资料DolphinDB分布式表DolphinDB数据库管理