This is an automated email from the ASF dual-hosted git repository.
shunping pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 4f934835059 Fix module-level side effects and global random seeding in
univariate ML anomaly tests (#39462)
4f934835059 is described below
commit 4f9348350592c56f5eada496df555f8dfd4497f5
Author: Yi Hu <[email protected]>
AuthorDate: Thu Jul 30 11:58:48 2026 -0400
Fix module-level side effects and global random seeding in univariate ML
anomaly tests (#39462)
* Remove top-level dataset generation, stdout printing, and global seeding
from perf_test.py by moving setup logic into PerfTest.setUpClass.
* Replace calls to global random.seed() in perf_test.py, mean_test.py,
quantile_test.py, and stdev_test.py with isolated random.Random instances.
This prevents state leakage and unneeded computation when modules are
imported during test collection.
---
.../apache_beam/ml/anomaly/univariate/mean_test.py | 8 ++--
.../apache_beam/ml/anomaly/univariate/perf_test.py | 51 ++++++++++++----------
.../ml/anomaly/univariate/quantile_test.py | 8 ++--
.../ml/anomaly/univariate/stdev_test.py | 8 ++--
4 files changed, 39 insertions(+), 36 deletions(-)
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 0c8888597eb..d0d7a62e4ce 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 @@ class LandmarkMeanTest(unittest.TestCase):
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 @@ class SlidingMeanTest(unittest.TestCase):
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 61067ef38da..9bb4004cf0e 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.mean import *
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 9f9402505b7..427c66a964b 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 @@ class LandmarkQuantileTest(unittest.TestCase):
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 @@ class SlidingQuantileTest(unittest.TestCase):
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 c26f16e3a1b..f29fdccd91e 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 @@ class LandmarkStdevTest(unittest.TestCase):
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 @@ class SlidingStdevTest(unittest.TestCase):
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)
