diff --git a/sdks/python/apache_beam/ml/anomaly/univariate/mean_test.py b/sdks/python/apache_beam/ml/anomaly/univariate/mean_test.py index 0c8888597ebe..d0d7a62e4ce1 100644 --- a/sdks/python/apache_beam/ml/anomaly/univariate/mean_test.py +++ b/sdks/python/apache_beam/ml/anomaly/univariate/mean_test.py @@ -70,13 +70,13 @@ def test_with_float64_max(self): def test_accuracy_fuzz(self): seed = int(time.time()) - random.seed(seed) + rng = random.Random(seed) print("Random seed: %d" % seed) for _ in range(10): numbers = [] for _ in range(5000): - numbers.append(random.randint(0, 1000)) + numbers.append(rng.randint(0, 1000)) with warnings.catch_warnings(record=False): warnings.simplefilter("ignore") @@ -140,13 +140,13 @@ def test_with_float64_max(self, tracker): def test_accuracy_fuzz(self): seed = int(time.time()) - random.seed(seed) + rng = random.Random(seed) print("Random seed: %d" % seed) for _ in range(10): numbers = [] for _ in range(5000): - numbers.append(random.randint(0, 1000)) + numbers.append(rng.randint(0, 1000)) t1 = IncSlidingMeanTracker(100) t2 = SimpleSlidingMeanTracker(100) diff --git a/sdks/python/apache_beam/ml/anomaly/univariate/perf_test.py b/sdks/python/apache_beam/ml/anomaly/univariate/perf_test.py index 61067ef38dab..9bb4004cf0e7 100644 --- a/sdks/python/apache_beam/ml/anomaly/univariate/perf_test.py +++ b/sdks/python/apache_beam/ml/anomaly/univariate/perf_test.py @@ -27,14 +27,6 @@ from apache_beam.ml.anomaly.univariate.quantile import * from apache_beam.ml.anomaly.univariate.stdev import * -seed_value_time = int(time.time()) -random.seed(seed_value_time) -print(f"{'Seed value':32s}{seed_value_time}") - -numbers = [] -for _ in range(50000): - numbers.append(random.randint(0, 1000)) - def run_tracker(tracker, numbers): for i in range(len(numbers)): @@ -42,39 +34,50 @@ def run_tracker(tracker, numbers): _ = tracker.get() -def print_result(tracker, number=10, repeat=5): - runtimes = timeit.repeat( - lambda: run_tracker(tracker, numbers), number=number, repeat=repeat) - mean = statistics.mean(runtimes) - sd = statistics.stdev(runtimes) - print(f"{tracker.__class__.__name__:32s}{mean:.6f} ± {sd:.6f}") +class PerfTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + seed_value_time = int(time.time()) + rng = random.Random(seed_value_time) + print(f"{'Seed value':32s}{seed_value_time}") + cls.numbers = [] + for _ in range(50000): + cls.numbers.append(rng.randint(0, 1000)) + + def print_result(self, tracker, number=10, repeat=5): + runtimes = timeit.repeat( + lambda: run_tracker(tracker, self.numbers), + number=number, + repeat=repeat) + mean = statistics.mean(runtimes) + sd = statistics.stdev(runtimes) + print(f"{tracker.__class__.__name__:32s}{mean:.6f} ± {sd:.6f}") -class PerfTest(unittest.TestCase): def test_mean_perf(self): print() - print_result(IncLandmarkMeanTracker()) - print_result(IncSlidingMeanTracker(100)) + self.print_result(IncLandmarkMeanTracker()) + self.print_result(IncSlidingMeanTracker(100)) # SimpleSlidingMeanTracker (numpy-based batch approach) is an order of # magnitude slower than other methods. To prevent excessively long test # runs, we reduce the number of repetitions. - print_result(SimpleSlidingMeanTracker(100), number=1) + self.print_result(SimpleSlidingMeanTracker(100), number=1) def test_stdev_perf(self): print() - print_result(IncLandmarkStdevTracker()) - print_result(IncSlidingStdevTracker(100)) + self.print_result(IncLandmarkStdevTracker()) + self.print_result(IncSlidingStdevTracker(100)) # Same as test_mean_perf, we reduce the number of repetitions here. - print_result(SimpleSlidingStdevTracker(100), number=1) + self.print_result(SimpleSlidingStdevTracker(100), number=1) def test_quantile_perf(self): print() with warnings.catch_warnings(record=False): warnings.simplefilter("ignore") - print_result(BufferedLandmarkQuantileTracker(0.5)) - print_result(BufferedSlidingQuantileTracker(100, 0.5)) + self.print_result(BufferedLandmarkQuantileTracker(0.5)) + self.print_result(BufferedSlidingQuantileTracker(100, 0.5)) # Same as test_mean_perf, we reduce the number of repetitions here. - print_result(SimpleSlidingQuantileTracker(100, 0.5), number=1) + self.print_result(SimpleSlidingQuantileTracker(100, 0.5), number=1) if __name__ == '__main__': diff --git a/sdks/python/apache_beam/ml/anomaly/univariate/quantile_test.py b/sdks/python/apache_beam/ml/anomaly/univariate/quantile_test.py index 9f9402505b70..427c66a964bc 100644 --- a/sdks/python/apache_beam/ml/anomaly/univariate/quantile_test.py +++ b/sdks/python/apache_beam/ml/anomaly/univariate/quantile_test.py @@ -68,13 +68,13 @@ def test_with_nan(self): def test_accuracy_fuzz(self): seed = int(time.time()) - random.seed(seed) + rng = random.Random(seed) print("Random seed: %d" % seed) def _accuracy_helper(): numbers = [] for _ in range(5000): - numbers.append(random.randint(0, 1000)) + numbers.append(rng.randint(0, 1000)) with warnings.catch_warnings(record=False): warnings.simplefilter("ignore") @@ -138,13 +138,13 @@ def test_with_nan(self, tracker): def test_accuracy_fuzz(self): seed = int(time.time()) - random.seed(seed) + rng = random.Random(seed) print("Random seed: %d" % seed) def _accuracy_helper(): numbers = [] for _ in range(5000): - numbers.append(random.randint(0, 1000)) + numbers.append(rng.randint(0, 1000)) t1 = BufferedSlidingQuantileTracker(100, 0.1) t2 = SimpleSlidingQuantileTracker(100, 0.1) diff --git a/sdks/python/apache_beam/ml/anomaly/univariate/stdev_test.py b/sdks/python/apache_beam/ml/anomaly/univariate/stdev_test.py index c26f16e3a1b7..f29fdccd91e6 100644 --- a/sdks/python/apache_beam/ml/anomaly/univariate/stdev_test.py +++ b/sdks/python/apache_beam/ml/anomaly/univariate/stdev_test.py @@ -66,13 +66,13 @@ def test_with_nan(self): def test_accuracy_fuzz(self): seed = int(time.time()) - random.seed(seed) + rng = random.Random(seed) print("Random seed: %d" % seed) for _ in range(10): numbers = [] for _ in range(5000): - numbers.append(random.randint(0, 1000)) + numbers.append(rng.randint(0, 1000)) t1 = IncLandmarkStdevTracker() t2 = SimpleSlidingStdevTracker(len(numbers)) @@ -135,13 +135,13 @@ def test_stdev_with_nan(self, tracker): def test_accuracy_fuzz(self): seed = int(time.time()) - random.seed(seed) + rng = random.Random(seed) print("Random seed: %d" % seed) for _ in range(10): numbers = [] for _ in range(5000): - numbers.append(random.randint(0, 1000)) + numbers.append(rng.randint(0, 1000)) t1 = IncSlidingStdevTracker(100) t2 = SimpleSlidingStdevTracker(100)