-
Notifications
You must be signed in to change notification settings - Fork 0
/
simple_tpch.py
192 lines (188 loc) · 9.11 KB
/
simple_tpch.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
"""
:Date: 2021-02-12
:Version: 0.1
:Authors: Patrick Erdelt
Perform TPC-H inspired benchmarks in a Kubernetes cluster.
This either profiles the imported data in several DBMS and compares some statistics, or runs the TPC-H queries.
Optionally monitoring is actived.
User can choose to detach the componenten of the benchmarking system, so that as much as possible is run inside a Kubernetes (K8s) cluster.
User can also choose some parameters like number of runs per query and configuration and request some resources.
"""
from bexhoma import *
from dbmsbenchmarker import *
import logging
import urllib3
import logging
import argparse
import time
from timeit import default_timer
import datetime
urllib3.disable_warnings()
logging.basicConfig(level=logging.ERROR)
if __name__ == '__main__':
description = """Perform TPC-H inspired benchmarks in a Kubernetes cluster.
This either profiles the imported data in several DBMS and compares some statistics, or runs the TPC-H queries.
Optionally monitoring is actived.
User can choose to detach the componenten of the benchmarking system, so that as much as possible is run inside a Kubernetes (K8s) cluster.
User can also choose some parameters like number of runs per query and configuration and request some resources.
"""
# argparse
parser = argparse.ArgumentParser(description=description)
parser.add_argument('mode', help='profile the import of TPC-H data, or run the TPC-H queries, or start DBMS and load data, or just start the DBMS', choices=['profiling', 'run', 'start', 'load'])
parser.add_argument('-db', '--debug', help='dump debug informations', action='store_true')
parser.add_argument('-c', '--connection', help='name of DBMS', default=None)
parser.add_argument('-cx', '--context', help='context of Kubernetes (for a multi cluster environment), default is current context', default=None)
parser.add_argument('-e', '--experiment', help='sets experiment code for continuing started experiment', default=None)
parser.add_argument('-d', '--detached', help='puts most of the experiment workflow inside the cluster', action='store_true')
parser.add_argument('-m', '--monitoring', help='activates monitoring', action='store_true')
parser.add_argument('-ms', '--max-sut', help='maximum number of parallel DBMS configurations, default is no limit', default=None)
parser.add_argument('-dt', '--datatransfer', help='activates datatransfer', action='store_true', default=False)
parser.add_argument('-md', '--monitoring-delay', help='time to wait [s] before execution of the runs of a query', default=10)
parser.add_argument('-nr', '--num-run', help='number of runs per query', default=1)
parser.add_argument('-nc', '--num-config', help='number of runs per configuration', default=1)
parser.add_argument('-ne', '--num-query-executors', help='comma separated list of number of parallel clients', default="1")
parser.add_argument('-sf', '--scaling-factor', help='scaling factor (SF)', default=1)
parser.add_argument('-t', '--timeout', help='timeout for a run of a query', default=180)
parser.add_argument('-rr', '--request-ram', help='request ram', default='16Gi')
parser.add_argument('-rc', '--request-cpu', help='request cpus', default='4')
parser.add_argument('-rct', '--request-cpu-type', help='request node having node label cpu=', default='')
parser.add_argument('-rg', '--request-gpu', help='request number of gpus', default=1)
parser.add_argument('-rgt', '--request-gpu-type', help='request node having node label gpu=', default='a100')
parser.add_argument('-rst', '--request-storage-type', help='request persistent storage of certain type', default=None, choices=[None, '', 'local-hdd', 'shared'])
parser.add_argument('-rss', '--request-storage-size', help='request persistent storage of certain size', default='10Gi')
parser.add_argument('-rnn', '--request-node-name', help='request a specific node', default=None)
# evaluate args
logger = logging.getLogger('bexhoma')
args = parser.parse_args()
# evaluate args
if args.debug:
logging.basicConfig(level=logging.DEBUG)
logging.basicConfig(level=logging.DEBUG)
# set parameter
monitoring = args.monitoring
connection = args.connection
mode = str(args.mode)
SF = str(args.scaling_factor)
timeout = int(args.timeout)
numRun = int(args.num_run)
num_experiment_to_apply = int(args.num_config)
cpu = str(args.request_cpu)
memory = str(args.request_ram)
cpu_type = str(args.request_cpu_type)
gpu_type = str(args.request_gpu_type)
gpus = str(args.request_gpu)
request_storage_type = args.request_storage_type
request_storage_size = args.request_storage_size
request_node_name = args.request_node_name
datatransfer = args.datatransfer
code = args.experiment
# set cluster
cluster = clusters.kubernetes(context=args.context)
cluster_name = cluster.contextdata['clustername']
if args.max_sut is not None:
cluster.max_sut = int(args.max_sut)
# set experiment
if code is None:
code = cluster.code
experiment = experiments.tpch(cluster=cluster, SF=SF, timeout=timeout, code=code, num_experiment_to_apply=num_experiment_to_apply)
if mode == 'run':
# we want all TPC-H queries
experiment.set_queries_full()
experiment.set_workload(
name = 'TPC-H Queries SF='+str(SF),
info = 'This experiment compares run time and resource consumption of TPC-H queries in different DBMS.'
)
else:
# we want to profile the import
experiment.set_queries_profiling()
experiment.set_workload(
name = 'TPC-H Data Profiling SF='+str(SF),
info = 'This experiment compares imported TPC-H data sets in different DBMS.'
)
if monitoring:
# we want to monitor resource consumption
experiment.set_querymanagement_monitoring(numRun=numRun, delay=10, datatransfer=datatransfer)
else:
# we want to just run the queries
experiment.set_querymanagement_quicktest(numRun=numRun, datatransfer=datatransfer)
# set resources for dbms
experiment.set_resources(
requests = {
'cpu': cpu,
'memory': memory,
'gpu': 0
},
limits = {
'cpu': 0,
'memory': 0
},
nodeSelector = {
'cpu': cpu_type,
'gpu': '',
})
if request_node_name is not None:
experiment.set_resources(
nodeSelector = {
'cpu': cpu_type,
'gpu': '',
'kubernetes.io/hostname': request_node_name
})
# persistent storage
#print(request_storage_type)
experiment.set_storage(
storageClassName = request_storage_type,
storageSize = request_storage_size,#'100Gi',
keep = False
)
cluster.start_dashboard()
# add configs
config = configurations.default(experiment=experiment, docker='MonetDB', configuration='MonetDB-{}'.format(cluster_name), alias='DBMS A')
config = configurations.default(experiment=experiment, docker='PostgreSQL', configuration='PostgreSQL-{}'.format(cluster_name), alias='DBMS D')
if connection is not None:
for c in reversed(range(len(experiment.configurations))):
if c.docker != connection:
del experiment.configurations
if args.mode == 'start':
experiment.start_sut()
elif args.mode == 'load':
# start all DBMS
experiment.start_sut()
# configure number of clients per config = 0
list_clients = []
# total time of experiment
experiment.add_benchmark_list(list_clients)
start = default_timer()
start_datetime = str(datetime.datetime.now())
print("Experiment starts at {} ({})".format(start_datetime, start))
# run workflow
experiment.work_benchmark_list()
# total time of experiment
end = default_timer()
end_datetime = str(datetime.datetime.now())
duration_experiment = end - start
print("Experiment ends at {} ({}): {}s total".format(end_datetime, end, duration_experiment))
else:
# configure number of clients per config
list_clients = args.num_query_executors.split(",")
if len(list_clients) > 0:
list_clients = [int(x) for x in list_clients]
experiment.add_benchmark_list(list_clients)
# total time of experiment
start = default_timer()
start_datetime = str(datetime.datetime.now())
print("Experiment starts at {} ({})".format(start_datetime, start))
# run workflow
experiment.work_benchmark_list()
# total time of experiment
end = default_timer()
end_datetime = str(datetime.datetime.now())
duration_experiment = end - start
print("Experiment ends at {} ({}): {}s total".format(end_datetime, end, duration_experiment))
##################
experiment.evaluate_results()
experiment.stop_benchmarker()
experiment.stop_sut()
cluster.restart_dashboard()
# OOM? exit code 137
#experiment.zip()
exit()