Anomaly Detection on Batch and Streaming Data using Apache Beam (Z-Score Method)

This notebook demonstrates how to perform anomaly detection on both batch and streaming data using the AnomalyDetection PTransform:

Batch Anomaly Detection: This section focuses on processing a static dataset. A synthetic univariate dataset containing outliers is generated. Subsequently, the AnomalyDetection PTransform, utilizing the Z-Score algorithm, is applied to identify and log the outliers.
Streaming Anomaly Detection with Concept Drift: This section simulates a real-time environment where the data distribution changes over time. A synthetic dataset incorporating both outliers and concept drift is published to a Pub/Sub topic. An Apache Beam pipeline is configured to:
- Read the streaming data from the input Pub/Sub topic.
- Apply the AnomalyDetection PTransform within a sliding window.
- Publish the enriched results (original data, anomaly scores, and labels) to an output Pub/Sub topic.
Finally, the labeled data points are visulaized in a series of plots to observe the detection performance in a streaming context with concept drift.

Preparation

To get started with this notebook, you'll need to install the Apache Beam Python SDK and its associated extras. Make sure your installation is version 2.64.0 or later.

 pip install 'apache_beam[interactive,gcp]>=2.64.0' --quiet

To proceed, import the essential modules: matplotlib, numpy, pandas, Beam, and others as needed.

# Import required dependencies for the notebook
import json
import os
import random
import threading
import time
import warnings
from typing import Any
from typing import Iterable
from typing import Tuple

import matplotlib.animation
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from IPython.display import HTML, Javascript
from google.api_core import retry
from google.api_core.exceptions import AlreadyExists
from google.cloud import pubsub_v1
from google.cloud.exceptions import NotFound

import apache_beam as beam
from apache_beam.io.gcp.pubsub import PubsubMessage
from apache_beam.ml.anomaly.base import AnomalyResult
from apache_beam.ml.anomaly.base import AnomalyPrediction
from apache_beam.ml.anomaly.detectors.zscore import ZScore
from apache_beam.ml.anomaly.transforms import AnomalyDetection
from apache_beam.ml.anomaly.univariate.mean import IncLandmarkMeanTracker
from apache_beam.ml.anomaly.univariate.stdev import IncLandmarkStdevTracker
from apache_beam.ml.anomaly.univariate.mean import IncSlidingMeanTracker
from apache_beam.ml.anomaly.univariate.stdev import IncSlidingStdevTracker
from apache_beam.options.pipeline_options import PipelineOptions

# Suppress logging warnings
os.environ["GRPC_VERBOSITY"] = "ERROR"
os.environ["GLOG_minloglevel"] = "2"
warnings.filterwarnings('ignore')

Next, replace <PROJECT_ID> with your Google Cloud project ID.

# GCP-related constant are listed below

# GCP project id
PROJECT_ID = 'apache-beam-testing'  # @param {type:'string'}

SUFFIX = str(random.randint(0, 10000))

# Pubsub topic and subscription for retrieving input data
INPUT_TOPIC = 'anomaly-input-' + SUFFIX
INPUT_SUB = INPUT_TOPIC + '-sub'

# Pubsub topic and subscription for collecting output result
OUTPUT_TOPIC = 'anomaly-output-' + SUFFIX
OUTPUT_SUB = OUTPUT_TOPIC + '-sub'

The last preparation step needs to authenticate your Google account and authorize your Colab notebook to access Google Cloud Platform (GCP) resources associated with the project set above.

from google.colab import auth
auth.authenticate_user(project_id=PROJECT_ID)

Part 1: Batch Processing

Generating Synthetic Data with Outliers

This process synthesizes a dataset (N=200) for anomaly detection. The generation consists of two key steps:

A base dataset is generated from a standard normal distribution (μ=0,σ=1).
Global outliers are introduced by replacing 1% of these points with values drawn from a normal distribution with a significant mean shift (μ=9,σ=1).

A fixed random seed is used to ensure reproducibility.

# The size of a segment in the synthetic data set.
seg_size = 200

# The ratio of global outliers introduced in the synthetic data set.
outlier_ratio = 0.01

# Random seed
seed = 1234

np.random.seed(seed)

# starting from a fixed distribution
data = np.random.normal(loc=0, scale=1, size=seg_size)

# adding outliers
outlier_idx = np.random.choice(len(data), size=int(outlier_ratio * len(data)), replace = False)

for idx in outlier_idx:
    data[idx] = np.random.normal(loc=9, scale=1, size=1).item()

df = pd.Series(data, name='f1')

Run the following code to visualize the dataset on a scatter plot.

plt.figure(figsize=(12, 4))
plt.xlim(0, 200)
plt.ylim(-10, 20)
plt.scatter(x=range(len(df)), y=df, s=10)

<matplotlib.collections.PathCollection at 0x7ef3027b49d0>

png

Run the Beam Pipeline on the Batch Data

The following Beam pipeline implements an anomaly detection workflow on batch data. It executes the following steps in sequence:

Ingest and Format: The pipeline begins by ingesting a collection of numerical data and converting each number into a beam.Row.
Key for Stateful Processing: A single global key is assigned to every element. This ensures all data is processed by a single instance of the downstream stateful transform.
Anomaly Detection: The AnomalyDetection PTransform is applied to the keyed data.
Log Outliers: A Filter transform inspects the prediction output from the detector, retaining only the elements flagged as anomalies (label == 1). These outlier records are then logged for inspection or downstream action.

options = PipelineOptions()
with beam.Pipeline(options=options) as p:
    _ = (p | beam.Create(data)
        | "Convert to Rows" >> beam.Map(lambda x: beam.Row(f1=float(x))).with_output_types(beam.Row)
        | beam.WithKeys(0)
        | AnomalyDetection(ZScore(features=["f1"],
                                  sub_stat_tracker=IncLandmarkMeanTracker(100),
                                  stdev_tracker=IncLandmarkStdevTracker(100)))
        | beam.Filter(lambda x: x[1].predictions[0].label == 1)
        | beam.LogElements()
    )

WARNING:apache_beam.options.pipeline_options:Discarding unparseable args: ['-f', '/root/.local/share/jupyter/runtime/kernel-ad4fe005-8e82-4549-bac6-63e8e4b4d9c1.json']
WARNING:apache_beam.options.pipeline_options:Discarding unparseable args: ['-f', '/root/.local/share/jupyter/runtime/kernel-ad4fe005-8e82-4549-bac6-63e8e4b4d9c1.json']
(0, AnomalyResult(example=Row(f1=9.544331108822645), predictions=[AnomalyPrediction(model_id='ZScore', score=8.672319197619325, label=1, threshold=3, info='', source_predictions=None)]))
(0, AnomalyResult(example=Row(f1=9.388712735779308), predictions=[AnomalyPrediction(model_id='ZScore', score=7.32926235264911, label=1, threshold=3, info='', source_predictions=None)]))

Part 2: Streaming Processing

Generating Synthetic Data with Concept Drift

This data generation process synthesizes a single data set (N=1000) composed of five distinct segments, each designed to simulate a specific distributional behavior or type of concept drift. After concatenating these segments, global outliers with a larger mean are injected to complete the dataset.

# The size of a segment in the synthetic data set. Each segment represents
# a collection of data points generated from either a fixed distribution
# or a drift from one distribution to another.
# The idea is inspired by https://github.com/yixiaoma666/SCAR.
seg_size = 200

# The ratio of global outliers introduced in the synthetic data set.
outlier_ratio = 0.01

# Random seed
seed = 1234

np.random.seed(seed)

# Starting from a fixed distribution
data_seg1 = np.random.normal(loc=0, scale=1, size=seg_size)

# A sudden change between data_seg1 and data_seg2
data_seg2 = np.random.normal(loc=3, scale=3, size=seg_size)

# A gradual change in data_seg3
data_seg3 = []
for i in range(seg_size):
    prob = 1 - 1.0 * i / seg_size
    c = np.random.choice(2, 1, p = [prob, 1 - prob])
    if c == 0:
        data_seg3.append(np.random.normal(loc=3, scale=3, size=1))
    else:
        data_seg3.append(np.random.normal(loc=0, scale=1, size=1))
data_seg3 = np.array(data_seg3).ravel()

# An incremental change in data_seg4
data_seg4 = []
for i in range(seg_size):
    loc = 0 + 3.0 * i / seg_size
    scale = 1 + 2.0 * i / seg_size
    data_seg4.append(np.random.normal(loc=loc, scale=scale, size=1))
data_seg4 = np.array(data_seg4).ravel()

# Back to a fixed distribution
data_seg5 = np.random.normal(loc=3, scale=3, size=seg_size)

# Combining all segements
data = np.concatenate((data_seg1, data_seg2, data_seg3, data_seg4, data_seg5))

# Adding global outliers
outlier_idx = np.random.choice(len(data), size=int(outlier_ratio * len(data)), replace = False)

for idx in outlier_idx:
    data[idx] = np.random.normal(loc=15, scale=1, size=1).item()

df = pd.Series(data, name='f1')