222 lines
8.5 KiB
Python
222 lines
8.5 KiB
Python
|
||
###################################################################
|
||
# Copyright (c) 2016 by TAOS Technologies, Inc.
|
||
# All rights reserved.
|
||
#
|
||
# This file is proprietary and confidential to TAOS Technologies.
|
||
# No part of this file may be reproduced, stored, transmitted,
|
||
# disclosed or used in any form or by any means other than as
|
||
# expressly provided by the written permission from Jianhui Tao
|
||
#
|
||
###################################################################
|
||
|
||
# -*- coding: utf-8 -*-
|
||
|
||
import sys
|
||
import taos
|
||
import threading
|
||
import time
|
||
from datetime import datetime
|
||
import numpy as np
|
||
|
||
|
||
class MyThread(threading.Thread):
|
||
|
||
def __init__(self, func, args=()):
|
||
super(MyThread, self).__init__()
|
||
self.func = func
|
||
self.args = args
|
||
|
||
def run(self):
|
||
self.result = self.func(*self.args)
|
||
|
||
def get_result(self):
|
||
try:
|
||
return self.result # 如果子线程不使用join方法，此处可能会报没有self.result的错误
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
class MetadataQuery:
|
||
def initConnection(self):
|
||
self.tables = 100
|
||
self.records = 10
|
||
self.numOfTherads = 5
|
||
self.ts = 1537146000000
|
||
self.host = "127.0.0.1"
|
||
self.user = "root"
|
||
self.password = "taosdata"
|
||
self.config = "/etc/taos"
|
||
self.conn = taos.connect(
|
||
self.host,
|
||
self.user,
|
||
self.password,
|
||
self.config)
|
||
|
||
def connectDB(self):
|
||
return self.conn.cursor()
|
||
|
||
def createStable(self):
|
||
print("================= Create stable meters =================")
|
||
cursor = self.connectDB()
|
||
cursor.execute("drop database if exists test")
|
||
cursor.execute("create database test")
|
||
cursor.execute("use test")
|
||
cursor.execute('''create table if not exists meters (ts timestamp, speed int) tags(
|
||
tgcol1 tinyint, tgcol2 smallint, tgcol3 int, tgcol4 bigint, tgcol5 float, tgcol6 double, tgcol7 bool, tgcol8 binary(20), tgcol9 nchar(20),
|
||
tgcol10 tinyint, tgcol11 smallint, tgcol12 int, tgcol13 bigint, tgcol14 float, tgcol15 double, tgcol16 bool, tgcol17 binary(20), tgcol18 nchar(20),
|
||
tgcol19 tinyint, tgcol20 smallint, tgcol21 int, tgcol22 bigint, tgcol23 float, tgcol24 double, tgcol25 bool, tgcol26 binary(20), tgcol27 nchar(20),
|
||
tgcol28 tinyint, tgcol29 smallint, tgcol30 int, tgcol31 bigint, tgcol32 float, tgcol33 double, tgcol34 bool, tgcol35 binary(20), tgcol36 nchar(20),
|
||
tgcol37 tinyint, tgcol38 smallint, tgcol39 int, tgcol40 bigint, tgcol41 float, tgcol42 double, tgcol43 bool, tgcol44 binary(20), tgcol45 nchar(20),
|
||
tgcol46 tinyint, tgcol47 smallint, tgcol48 int, tgcol49 bigint, tgcol50 float, tgcol51 double, tgcol52 bool, tgcol53 binary(20), tgcol54 nchar(20))''')
|
||
cursor.close()
|
||
|
||
def createTablesAndInsertData(self, threadID):
|
||
cursor = self.connectDB()
|
||
cursor.execute("use test")
|
||
base = threadID * self.tables
|
||
|
||
tablesPerThread = int(self.tables / self.numOfTherads)
|
||
for i in range(tablesPerThread):
|
||
cursor.execute(
|
||
'''create table t%d using meters tags(
|
||
%d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d',
|
||
%d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d',
|
||
%d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d',
|
||
%d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d',
|
||
%d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d',
|
||
%d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d')''' %
|
||
(base + i + 1, (base + i) %
|
||
100, (base + i) %
|
||
10000, (base + i) %
|
||
1000000, (base + i) %
|
||
100000000, (base + i) %
|
||
100 * 1.1, (base + i) %
|
||
100 * 2.3, (base + i) %
|
||
2, (base + i) %
|
||
100, (base + i) %
|
||
100, (base + i) %
|
||
100, (base + i) %
|
||
10000, (base + i) %
|
||
1000000, (base + i) %
|
||
100000000, (base + i) %
|
||
100 * 1.1, (base + i) %
|
||
100 * 2.3, (base + i) %
|
||
2, (base + i) %
|
||
100, (base + i) %
|
||
100, (base + i) %
|
||
100, (base + i) %
|
||
10000, (base + i) %
|
||
1000000, (base + i) %
|
||
100000000, (base + i) %
|
||
100 * 1.1, (base + i) %
|
||
100 * 2.3, (base + i) %
|
||
2, (base + i) %
|
||
100, (base + i) %
|
||
100, (base + i) %
|
||
100, (base + i) %
|
||
10000, (base + i) %
|
||
1000000, (base + i) %
|
||
100000000, (base + i) %
|
||
100 * 1.1, (base + i) %
|
||
100 * 2.3, (base + i) %
|
||
2, (base + i) %
|
||
100, (base + i) %
|
||
100, (base + i) %
|
||
100, (base + i) %
|
||
10000, (base + i) %
|
||
1000000, (base + i) %
|
||
100000000, (base + i) %
|
||
100 * 1.1, (base + i) %
|
||
100 * 2.3, (base + i) %
|
||
2, (base + i) %
|
||
100, (base + i) %
|
||
100, (base + i) %
|
||
100, (base + i) %
|
||
10000, (base + i) %
|
||
1000000, (base + i) %
|
||
100000000, (base + i) %
|
||
100 * 1.1, (base + i) %
|
||
100 * 2.3, (base + i) %
|
||
2, (base + i) %
|
||
100, (base + i) %
|
||
100))
|
||
for j in range(self.records):
|
||
cursor.execute(
|
||
"insert into t%d values(%d, %d)" %
|
||
(base + i + 1, self.ts + j, j))
|
||
cursor.close()
|
||
|
||
def queryWithTagId(self, threadId, tagId, queryNum):
|
||
print("---------thread%d start-----------" % threadId)
|
||
query = '''select tgcol1, tgcol2, tgcol3, tgcol4, tgcol5, tgcol6, tgcol7, tgcol8, tgcol9,
|
||
tgcol10, tgcol11, tgcol12, tgcol13, tgcol14, tgcol15, tgcol16, tgcol17, tgcol18,
|
||
tgcol19, tgcol20, tgcol21, tgcol22, tgcol23, tgcol24, tgcol25, tgcol26, tgcol27,
|
||
tgcol28, tgcol29, tgcol30, tgcol31, tgcol32, tgcol33, tgcol34, tgcol35, tgcol36,
|
||
tgcol37, tgcol38, tgcol39, tgcol40, tgcol41, tgcol42, tgcol43, tgcol44, tgcol45,
|
||
tgcol46, tgcol47, tgcol48, tgcol49, tgcol50, tgcol51, tgcol52, tgcol53, tgcol54
|
||
from meters where tgcol{id} > {condition}'''
|
||
latancy = []
|
||
cursor = self.connectDB()
|
||
cursor.execute("use test")
|
||
for i in range(queryNum):
|
||
startTime = time.time()
|
||
cursor.execute(query.format(id=tagId, condition=i))
|
||
cursor.fetchall()
|
||
latancy.append((time.time() - startTime))
|
||
print("---------thread%d end-----------" % threadId)
|
||
return latancy
|
||
|
||
def queryData(self, query):
|
||
cursor = self.connectDB()
|
||
cursor.execute("use test")
|
||
|
||
print("================= query tag data =================")
|
||
startTime = datetime.now()
|
||
cursor.execute(query)
|
||
cursor.fetchall()
|
||
endTime = datetime.now()
|
||
print(
|
||
"Query time for the above query is %d seconds" %
|
||
(endTime - startTime).seconds)
|
||
|
||
cursor.close()
|
||
# self.conn.close()
|
||
|
||
|
||
if __name__ == '__main__':
|
||
|
||
t = MetadataQuery()
|
||
t.initConnection()
|
||
|
||
latancys = []
|
||
threads = []
|
||
tagId = 1
|
||
queryNum = 1000
|
||
for i in range(t.numOfTherads):
|
||
thread = MyThread(t.queryWithTagId, args=(i, tagId, queryNum))
|
||
threads.append(thread)
|
||
thread.start()
|
||
for i in range(t.numOfTherads):
|
||
threads[i].join()
|
||
latancys.extend(threads[i].get_result())
|
||
print("Total query: %d" % (queryNum * t.numOfTherads))
|
||
print(
|
||
"statistic(s): mean= %f, P50 = %f, P75 = %f, P95 = %f, P99 = %f" %
|
||
(sum(latancys) /
|
||
(
|
||
queryNum *
|
||
t.numOfTherads),
|
||
np.percentile(
|
||
latancys,
|
||
50),
|
||
np.percentile(
|
||
latancys,
|
||
75),
|
||
np.percentile(
|
||
latancys,
|
||
95),
|
||
np.percentile(
|
||
latancys,
|
||
99)))
|