我曾经 influxDB-Python
插入从数据库读取的大量数据 Redis-Stream
. 因为redis stream并设置maxlen=600,数据以 100ms
,我需要保留它的所有数据。因此,我读取并将其传输到influxdb(我不知道什么是更好的数据库),但只使用批插入⌈计数/批量大小⌉ 在每个批大小的末尾,数据段似乎都被覆盖。以下代码
import redis
from apscheduler.schedulers.blocking import BlockingScheduler
import time
import datetime
import os
import struct
from influxdb import InfluxDBClient
def parse(datas):
ts,data = datas
w_json = {
"measurement": 'sensor1',
"fields": {
"Value":data[b'Value'].decode('utf-8')
"Count":data[b'Count'].decode('utf-8')
}
}
return w_json
def archived_data(rs,client):
results= rs.xreadgroup('group1', 'test', {'test1': ">"}, count=600)
if(len(results)!=0):
print("len(results[0][1]) = ",len(results[0][1]))
datas = list(map(parse,results[0][1]))
client.write_points(datas,batch_size=300)
print('insert success')
else:
print("No new data is generated")
if __name__=="__main__":
try:
rs = redis.Redis(host="localhost", port=6379, db=0)
rs.xgroup_destroy("test1", "group1")
rs.xgroup_create('test1','group1','0-0')
except Exception as e:
print("error = ",e)
try:
client = InfluxDBClient(host="localhost", port=8086,database='test')
except Exception as e:
print("error = ", e)
try:
sched = BlockingScheduler()
sched.add_job(test1, 'interval', seconds=60,args=[rs,client])
sched.start()
except Exception as e:
print(e)
XDB的数据如下所示
> select count(*) from sensor1;
name: sensor1
time count_Count count_Value
---- ----------- -----------
0 6 6
> select count(*) from sensor1;
name: sensor1
time count_Count count_Value
---- ----------- -----------
0 8 8
> select Count from sensor1;
name: sensor1
time Count
---- -----
1594099736722564482 00000310
1594099737463373188 00000610
1594099795941527728 00000910
1594099796752396784 00001193
1594099854366369551 00001493
1594099855120826270 00001777
1594099913596094653 00002077
1594099914196135122 00002361
为什么数据看起来会被覆盖,我如何解析它以一次插入所有数据?
如果你能告诉我怎么解决的话,我会很感激的?
1条答案
按热度按时间rmbxnbpk1#
您能提供更多关于您希望存储在influxdb中的数据结构的细节吗?不过,我希望下面的信息能对你有所帮助。
在XDB中,timestamp+标记是唯一的(即,具有相同标记值和timestamp的两个数据点不能存在)。与sql不同,xdb不会抛出唯一的约束冲突,它用传入数据覆盖现有数据。您的数据似乎没有标记,因此如果某个传入数据的时间戳已经存在于influxdb中,那么xdb将覆盖现有数据