DolphinDB API：Python
实时流数据接入是指将数据从数据源实时写入 DolphinDB 中，供后续进行清洗、计算等使用。本文档介绍如何使用 Python API 把数据写入 DolphinDB。
按照上游发布端数据产生频率大小推荐不同的写入接口：
-
当数据实时写入频率小于100条每秒时，推荐使用 tableInsert 同步写入接口，其写入延时相对其它接口是最低的
-
当数据实时写入频率大于100条每秒时，推荐使用 MultithreadedTableWriter 异步写入接口，其带来的吞吐量和写入延时综合效益最高
下面将介绍两种场景下推荐函数的使用方法和基准性能。
如果本文档中遇到 DolphinDB Python API 使用相关问题，参考：Python API
tableInsert 同步写入接口
当数据实时写入频率小于 100 条每秒时，推荐使用 tableInsert 同步接口写入数据。
本文档提供 10 列数据的写入示例和基准性能。
step1：创建共享普通流数据表
import dolphindb as ddb import numpy as np import pandas as pd import random s = ddb.session() s.connect(host="localhost", port=8848, userid="admin", password="123456") script = """ name = `c1Time`c2Symbol`c3Int`c4Int`c5Long`c6Long`c7Float`c8Float`c9Double`c10ArrayInt type = [TIMESTAMP, SYMBOL, INT, INT, LONG, LONG, FLOAT, FLOAT, DOUBLE, INT[]] share(table=streamTable(20000:0, name, type), sharedName="sharedMinBar") """ s.run(script)
注意：生产环境推荐使用共享持久化流数据表。
step2：tableInsert 写入数据
def createData():
c1Time = [pd.Timestamp.now()]
c2Symbol = ['000001']
c3Int = [random.randint(0, 10)]
c4Int = [random.randint(0, 10)]
c5Long = [random.randint(0, 1000)]
c6Long = [random.randint(0, 1000)]
c7Float = [random.uniform(0, 1)]
c8Float = [random.uniform(0, 1)]
c9Double = [random.uniform(0, 1)]
c10ArrayInt = [np.random.randint(10,size=10)]
return [c1Time, c2Symbol, c3Int, c4Int, c5Long, c6Long, c7Float, c8Float, c9Double, c10ArrayInt]
if __name__ == "__main__":
inertRows = 10000
for i in range(inertRows):
insertData = createData()
s.run("tableInsert{sharedMinBar}",insertData)
step3：查询表中数据
pd = s.run("select * from sharedMinBar limit 10")
返回：
step4：删除流数据表并关闭当前会话
script = """
dropStreamTable(tableName="sharedMinBar")
"""
s.run(script)
s.close()
基准性能
tableInsert 同步写入接口基准性能测试代码请参考附录 B，测试结果如下：
|
指标 |
性能 |
|---|---|
| 最小单次插入耗时 | 0.31 ms |
| 最大单次插入耗时 | 1.24 ms |
| 平均单次插入耗时 | 0.69 ms |
单次插入耗时统计图如下：
从基准性能测试结果显示：
-
DolphinDB Python API 的 tableInsert 同步写入接口，实时逐条写入共享流数据表的平均耗时是 0.69 毫秒
-
DolphinDB Python API 的 tableInsert 同步写入接口的稳定性较好，波动较小，平均单条数据写入的最大耗时小于 2 毫秒
MultithreadedTableWriter 异步写入接口
当数据写入频率大于 100 条每秒时，推荐使用 MultithreadedTableWriter（MTW） 函数写入数据。
如下图所示，MultithreadedTableWriter 函数把 Python 端实时产生的数据先通过 insert 线程写入
MTW 消息队列，同时 append 线程会批量取出消息队列的数据，自动转换数据类型后把数据写入 DolphinDB。
下面为 MultithreadedTableWriter 使用示例。
-
逐笔成交表
import numpy as np import pandas as pd import dolphindb as ddb import random import datetime s = ddb.session() s.connect(host="localhost", port=8848, userid="admin", password="123456") writer = ddb.MultithreadedTableWriter("localhost", 8848, "admin", "123456","","tglobal",False,False,[],10,0,2,"SecurityID") start = datetime.datetime.now() for i in range(100): res = writer.insert(random.randint(1,100),random.randint(1,100),str(i),random.randint(1,100),random.randint(1,100),str(i),str(i),random.random(),random.randint(1,100),str(i),np.datetime64('2021-01-04T09:30:02.000'),9,random.randint(1,100),random.randint(1,100),random.random(),str(i),random.randint(1,100),str(i),str(i)) writer.waitForThreadCompletion() print(writer.getStatus()) #output ''' errorCode : None errorInfo : isExiting : True sentRows : 10000 unsentRows : 0 sendFailedRows: 0 threadStatus : threadId sentRows unsentRows sendFailedRows 0 0 0 0 2474632960 4994 0 0 2483025664 5006 0 0 <dolphindb.session.MultithreadedTableWriterStatus object at 0x7f0ca9f0e8d0> '''
-
arrayVector 类型快照表
writer = ddb.MultithreadedTableWriter("localhost", 8892, "admin", "123456","","tglobal",False,False,[],10,0,2,"SecurityID") for i in range(10000): res = writer.insert(str(i),np.datetime64('2021-01-04T09:30:02.000'),random.random(),random.random(),random.random(),random.random(),random.random(),random.randint(1,100),random.random(),"OCALL",np.random.rand(10),np.random.randint(10,size=10),np.random.randint(10,size=10),np.random.randint(10,size=10),np.random.rand(10),np.random.randint(10,size=10),np.random.randint(10,size=10),np.random.randint(10,size=10),random.randint(1,100),random.random(),random.randint(1,100),random.randint(1,100),random.random(),random.random(),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.random(),random.randint(1,100),random.randint(1,100),random.random(),random.randint(1,100),random.randint(1,100),random.random(),random.randint(1,100),random.randint(1,100),random.random()) writer.waitForThreadCompletion() print(writer.getStatus()) #output ''' errorCode : None errorInfo : isExiting : True sentRows : 10000 unsentRows : 0 sendFailedRows: 0 threadStatus : threadId sentRows unsentRows sendFailedRows 0 0 0 0 2474632960 4994 0 0 2466240256 5006 0 0 <dolphindb.session.MultithreadedTableWriterStatus object at 0x7f0cb032e190> '''
-
multicolumn 类型快照表
writer = ddb.MultithreadedTableWriter("localhost", 8892, "admin", "123456","","tglobal",False,False,[],10,0,2,"SecurityID") for i in range(10000): res = writer.insert(str(i),np.datetime64('2021-01-04T09:30:02.000'),random.random(),random.random(),random.random(),random.random(),random.random(),random.randint(1,100),random.random(),str(i),random.random(),random.random(),random.random(),random.random(),random.random(),random.random(),random.random(),random.random(),random.random(),random.random(),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.random(),random.random(),random.random(),random.random(),random.random(),random.random(),random.random(),random.random(),random.random(),random.random(),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.random(),random.randint(1,100),random.randint(1,100),random.random(),random.random(),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.randint(1,100),random.random(),random.randint(1,100),random.randint(1,100),random.random(),random.randint(1,100),random.randint(1,100),random.random(),random.randint(1,100),random.randint(1,100),random.random()) writer.waitForThreadCompletion() print(writer.getStatus()) #output ''' errorCode : None errorInfo : isExiting : True sentRows : 10000 unsentRows : 0 sendFailedRows: 0 threadStatus : threadId sentRows unsentRows sendFailedRows 0 0 0 0 2474632960 4994 0 0 2466240256 5006 0 0 <dolphindb.session.MultithreadedTableWriterStatus object at 0x7f0ca9b37fd0> '''
附录 A-测试环境
| 硬件名称 | 配置信息 |
|---|---|
| CPU | Intel(R) Xeon(R) Silver 4216 CPU @ 2.10GHz |
| 内存 | 32 GB |
| 磁盘 | SSD 500 GB |
| 软件名称 | 版本信息 |
|---|---|
| 操作系统 | CentOS Linux 7 (Core) |
| 内存 | 3.10.0-1160.el7.x86_64 |
| DolphinDB server | 2.00.10.9 2023.12.09 |
| DolphinDB Python API | 1.30.22.6 |
| Python | 3.11.5 |
| Numpy | 1.24.4 |
| Pandas | 2.1.1 |
附录 B-tableInsert 同步写入接口基准性能测试代码
import dolphindb as ddb
import numpy as np
import pandas as pd
import random
#模拟数据
def createData():
c1Time = [pd.Timestamp.now()]
c2Symbol = ['000001']
c3Int = [random.randint(0, 10)]
c4Int = [random.randint(0, 10)]
c5Long = [random.randint(0, 1000)]
c6Long = [random.randint(0, 1000)]
c7Float = [random.uniform(0, 1)]
c8Float = [random.uniform(0, 1)]
c9Double = [random.uniform(0, 1)]
c10ArrayInt = [np.random.randint(10,size=10)]
return [c1Time, c2Symbol, c3Int, c4Int, c5Long, c6Long, c7Float, c8Float, c9Double, c10ArrayInt]
if __name__ == "__main__":
#与DolphinDB建立会话
s = ddb.session()
s.connect(host="localhost", port=8200, userid="admin", password="123456")
#创建共享普通流数据表
script = """
name = `c1Time`c2Symbol`c3Int`c4Int`c5Long`c6Long`c7Float`c8Float`c9Double`c10ArrayInt
type = [TIMESTAMP, SYMBOL, INT, INT, LONG, LONG, FLOAT, FLOAT, DOUBLE, INT[]]
share(table=streamTable(20000:0, name, type), sharedName="sharedMinBar")
"""
s.run(script)
#测试写入数据行数
inertRows = 10000
#延时统计
minCostTime = 1000.0
maxCostTime = 0.0
totalCostTime = 0.0
count = 0
for i in range(inertRows):
insertData = createData()
startTime = pd.Timestamp.now()
s.run("tableInsert{sharedMinBar}",insertData)
delta = (pd.Timestamp.now() - startTime) / np.timedelta64(1, 'ms')
if(delta < minCostTime):
minCostTime = delta
if(delta > maxCostTime):
maxCostTime = delta
totalCostTime += delta
count += 1
#每隔一段时间打印写入性能测试结果
if(count %10000 == 0):
print(f"min: {minCostTime} ms")
print(f"max: {maxCostTime} ms")
print("count:", count)
print(f"totalCostTime: {totalCostTime} ms")
print(f"avg: {totalCostTime / count} ms")
#删除流数据表
script = """
dropStreamTable(tableName="sharedMinBar")
"""
s.run(script)
#关闭当前会话
s.close()