DolphinDB 实现边到云的实时数据汇聚

本次设备使用两台台服务器，一台模拟云设备，用来接收同步数据，一台模拟边缘设备，用来生成需要同步的数据。设备的硬件资源以及 DolphinDB 的配置如下表3-1，3-2所示。

主机	说明
10.0.0.80	模拟云设备
10.0.0.81	模拟边设备1

• 硬件资源

模拟环境	数量	实际环境	型号/配置/软件名称/软件版本号
云端主机	1	物理服务器1台	OS：Centos 7.6
			CPU：Intel(R) Xeon(R) Silver 4214 CPU @ 2.20GHz（48逻辑CPU核数或超线程数）
			Mem：188 GB
			DolphinDB: 3.00.0.1
边端主机	1	物理服务器1台	Disk：2T SSD * 2 500M/s ;2T HDD*1 250M/s
			Net：100Mbps
			DolphinDB: 3.00.0.1

• DolphinDB 配置信息

云端设备配置信息如下所示：

localSite=localhost:8848:local8848
mode=single
maxMemSize=32
maxConnections=512
workerNum=4
localExecutors=3
maxBatchJobWorker=4
dataSync=1
OLAPCacheEngineSize=2
TSDBCacheEngineSize=1
newValuePartitionPolicy=add
maxPubConnections=64
subExecutors=4
perfMonitoring=true
lanCluster=0
subThrottle=1
persistenceWorkerNum=1
maxPubQueueDepthPerSite=100000000

为模拟边端设备资源受限的情况，本次测试边端设备配置为 4C8G，边端设备配置信息如下所示：

localSite=localhost:8848:local8848
mode=single
maxMemSize=8
maxConnections=512
workerNum=4
localExecutors=3
maxBatchJobWorker=4
dataSync=1
OLAPCacheEngineSize=2
TSDBCacheEngineSize=2
newValuePartitionPolicy=add
maxPubConnections=64
subExecutors=4
perfMonitoring=true
lanCluster=0
subThrottle=1
persistenceWorkerNum=1

本方案测试 DolphinDB 在不同数据量下的同步性能，以及内存，CPU 还有网络带宽的消耗情况 。测试范围涵盖从4,800条数据到14,400,000条数据的同步，适应大部分实际应用场景。此外，方案还测试了在使用 remoteRun 函数时，不同压缩方式（ zstd 和 lz4）下的资源消耗和传输效率，以确定不同条件下的最优方案。

数据量（行）	压缩方式	耗时（毫秒）	平均每秒同步量(条/S)	CPU使用	内存使用（G）	网络传输（MB/s）
4,800	lz4	60	80,000	103%	0.6	0.04
	zstd	62	77,419	105%	0.6	0.04
	无	74	64,864	105%	0.6	0.06
240,000	lz4	1,883	127,456	120%	0.7	1.22
	zstd	1,720	139,535	120%	0.7	0.94
	无	2,853	84,122	120%	0.7	2.12
1,920,000	lz4	14,384	133,482	140%	1.1	9.5
	zstd	13,208	145,366	140%	1.1	7.93
	无	22,362	85,859	145%	1.1	11.5
14,400,000	lz4	108,529	132,683	140%	1.7	10.1
	zstd	98,080	146,819	140%	1.8	8.85
	无	167,579	85,930	145%	1.3	11.4

DolphinDB 可以在较大数据量下的情况下实现高效的同步，同时保持较低的内存和CPU资源消耗，能够有效节约云端资源。

从表格中可以明显看到，带压缩传输的效率以及同步速度都高于未压缩的情况。比较 lz4 和zstd 两种压缩方式可以发现，zstd 压缩的压缩比更高，但是解压时间也更长。从表中可以看到，在小数据量的情况下，由于解压缩时间过长，因此 zstd 传输效率不如 lz4 传输；数据量越大，zstd 的传输效率相比于 lz4 越快。在小数据量（4800条数据）下，各个传输方式的传输效率基本相同，且其它资源消耗也基本相同。

对比无压缩的方式与 lz4 压缩的方式，网络传输速率的峰值已经基本达到本实验设备的带宽上限。想要提升同步速度，需要提升资源分配，提高带宽配置。无压缩的方式由于完整的传输所有数据，所以消耗在传输过程中的时间更高，消耗的时间高于压缩与解压缩的时间，因此速度慢于有压缩的方式。

对于 zstd 方式，限制其同步速度的主要原因不是网络带宽，而是订阅分发速度。因此想要提升这种方式的同步速度，需要升级 DolphinDB 的配置。

具体的实现细节以及测试过程在附录说明。

1. lz4 压缩

首先继续按照上面的数据库格式创建存放数据的数据库，在发送端创建流表并订阅自己的流表并且常见 handler 函数，在流表有新数据写入时触发函数，函数的主要功能就是将流表中的数据，利用 remoterRun 函数发送到接收端并写入到接收端的数据库中，发送的数据使用 compress 函数进行压缩并在接收端使用 decompress 函数解压后写入数据库，由于订阅流表会在产生新数据过程中分段发送，因此发送耗时也需要分段记录，为此在接收端创建了保存每一段订阅数据发送耗时的数据表。相关部分主要代码如下。

云端（接收端）

login("admin","123456")
share streamTable(1:0,`sourceDevice`startTime`endTime`cost`rowCount,[INT,TIMESTAMP,TIMESTAMP,DOUBLE,INT]) as timeCost

colNames = `device_id`error`detect_time`temp`humidity`temp_max`humidity_max`trace`position`Point_one`point_two`point_three`point_four`point_five`point_six`point_seven`point_eight`point_nine`point_ten`responsible`create_time`rob_type`rob_sn`rob_create`rob_acc1`rob_acc2`rob_acc3`rob_acc4`rob_acc5
colTypes = [SYMBOL,BOOL,TIMESTAMP,DOUBLE,DOUBLE,DOUBLE,DOUBLE,INT,SYMBOL,BOOL,BOOL, BOOL, BOOL, BOOL, BOOL, BOOL, BOOL, BOOL, BOOL, SYMBOL, TIMESTAMP, INT, SYMBOL, TIMESTAMP, DOUBLE, DOUBLE, DOUBLE, DOUBLE, DOUBLE]

if(existsDatabase("dfs://Robot")){dropDatabase("dfs://Robot");}
data= table(1:0, colNames, colTypes)
db = database(directory = "dfs://Robot",partitionType = VALUE, partitionScheme = [today()],engine=`TSDB)
createPartitionedTable(dbHandle=db, table=data, tableName = `data, partitionColumns = `detect_time,sortColumns = `device_id`detect_time)

边端（发送端）

def pushData(ip,port,compressMethod,msg){
        pubNodeHandler1=xdb(ip,port,`admin,123456)
        StartTime=now()
        table1=select * from msg;
        y=compress(table1,compressMethod)
        remoteRun(pubNodeHandler1,'def insertTableToDatabase(y,StartTime){
                                        
                                        pt=loadTable("dfs://Robot",`data)
                                        pt.append!(decompress(y))
                                        endTime=now()
                                        duration=endTime-StartTime
                                        tmp = table([0] as c1,[StartTime] as c2,[endTime] as c3,[duration] as c4,[y.size()] as c5)
                                        timeCost.append!(tmp)
                        }')
       
        remoteRun(pubNodeHandler1,'insertTableToDatabase',y,StartTime)
}
ip = "xx.xx.xx.xx"
port = xxxx
compressMethod = "lz4"
subscribeTable(,`data,`pushData,0,pushData{ip,port,compressMethod},msgAsTable=true,batchSize=500000,throttle=1)

为了对比在不同传输数据量下不同方案的性能，通过不断改变生成的数据量并测量不同情况下的各个资源的消耗，最终结果汇总如下。

随着传输数据量的增加，网络带宽消耗明显增加，内存消耗以及 cpu 消耗小幅增加，其中网络传输基本到达网络传输瓶颈。可以看到网络传输速率已经基本达到本实验设备的带宽上限，想要提升同步速度，需要提升资源分配，提高带宽配置。

2.zstd 压缩

将压缩格式改为 zstd。compressMethod="zstd"

3. 无压缩

这部分主要是与带压缩的传输形成对比，将发送端代码替换为如下代码：

def pushDataNoCompress(ip,port,msg){
        pubNodeHandler1=xdb(ip,port,`admin,123456)
        StartTime=now()
        remoteRun(pubNodeHandler1,'def insertTableToDatabase(y,StartTime){
                                        
                                        pt=loadTable("dfs://Robot",`data)
                                        pt.append!(y)
                                        endTime=now()
                                        duration=endTime-StartTime
                                        tmp = table([0] as c1,[StartTime] as c2,[endTime] as c3,[duration] as c4,[y.size()] as c5)
                                        timeCost.append!(tmp)
                        }')
       
        remoteRun(pubNodeHandler1,'insertTableToDatabase',msg,StartTime)
}
ip = "xx.xx.xx.xx"
port = xxxx
subscribeTable(,`data,`pushData,0,pushDataNoCompress{ip,port},msgAsTable=true,batchSize=500000,throttle=1)

附件中包含云边数据同步方案测试脚本 : 云边数据同步.zip