-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathprofile_ddsketch.py
More file actions
50 lines (39 loc) · 1.47 KB
/
profile_ddsketch.py
File metadata and controls
50 lines (39 loc) · 1.47 KB
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
import cProfile
import pstats
import io
import numpy as np # Using numpy for faster random data generation
from QuantileFlow.ddsketch.core import DDSketch
def run_sketch_operations():
"""Runs typical DDSketch operations for profiling."""
print("Initializing DDSketch...")
sketch = DDSketch(relative_accuracy=0.01)
num_values = 1_000_000
print(f"Inserting {num_values} random values...")
# Generate random data more efficiently with numpy
data = np.random.rand(num_values) * 1000
for value in data:
sketch.insert(value)
quantiles_to_compute = [0.5, 0.9, 0.99, 0.999]
print(f"Computing quantiles: {quantiles_to_compute}...")
for q in quantiles_to_compute:
try:
quantile_value = sketch.quantile(q)
print(f"Quantile({q}): {quantile_value}")
except ValueError as e:
print(f"Error computing quantile {q}: {e}")
print("Profiling complete.")
def profile():
"""Profiles the run_sketch_operations function."""
profiler = cProfile.Profile()
profiler.enable()
run_sketch_operations()
profiler.disable()
s = io.StringIO()
# Sort stats by cumulative time spent in the function and its callees
sortby = pstats.SortKey.CUMULATIVE
ps = pstats.Stats(profiler, stream=s).sort_stats(sortby)
ps.print_stats()
print("\n--- cProfile Results (Sorted by Cumulative Time) ---")
print(s.getvalue())
if __name__ == "__main__":
profile()