 Run in Google Colab
|
 View source on GitHub
|
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' --quietTo 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>
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
AnomalyDetectionPTransform is applied to the keyed data.Log Outliers: A
Filtertransform 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')
Run the following code to visualize the dataset on a scatter plot.
plt.figure(figsize=(12, 4))
plt.xlim(0, 1000)
plt.ylim(-10, 20)
plt.scatter(x=range(len(df)), y=df, s=2)
<matplotlib.collections.PathCollection at 0x7ef2fe753e90>
Setting Up Input/Output Pubsubs
Use the following code to create pubsub topics for input and output.
def create_topic_if_not_exists(project_id:str, topic_name:str, enable_message_ordering=False):
if enable_message_ordering:
# see https://cloud.google.com/pubsub/docs/ordering#python for details.
publisher_options = pubsub_v1.types.PublisherOptions(enable_message_ordering=True)
# Sending messages to the same region ensures they are received in order
client_options = {"api_endpoint": "us-east1-pubsub.googleapis.com:443"}
publisher = pubsub_v1.PublisherClient(publisher_options=publisher_options, client_options=client_options)
else:
publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project_id, topic_name)
try:
topic = publisher.create_topic(request={"name": topic_path})
print(f"Created topic: {topic.name}")
except AlreadyExists:
print(f"Topic {topic_path} already exists.")
return publisher
def create_subscription_if_not_exists(project_id:str, subscription_name:str, topic_name: str, enable_message_ordering=False):
topic_path = pubsub_v1.PublisherClient.topic_path(project_id, topic_name)
subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(project_id, subscription_name)
try:
subscription = subscriber.create_subscription(
request={"name": subscription_path, "topic": topic_path, "enable_message_ordering": enable_message_ordering}
)
print(f"Created subscription: {subscription.name}")
except AlreadyExists:
print(f"Subscription {subscription_path} already exists.")
return subscriber
# for input data
input_publisher = create_topic_if_not_exists(PROJECT_ID, INPUT_TOPIC, True)
create_subscription_if_not_exists(PROJECT_ID, INPUT_SUB, INPUT_TOPIC, True)
# for output data
create_topic_if_not_exists(PROJECT_ID, OUTPUT_TOPIC)
output_subscriber = create_subscription_if_not_exists(PROJECT_ID, OUTPUT_SUB, OUTPUT_TOPIC)
Created topic: projects/apache-beam-testing/topics/anomaly-input-9625 Created subscription: projects/apache-beam-testing/subscriptions/anomaly-input-9625-sub Created topic: projects/apache-beam-testing/topics/anomaly-output-9625 Created subscription: projects/apache-beam-testing/subscriptions/anomaly-output-9625-sub
Publishing Input to Pub/Sub
To simulate a live data stream without blocking the execution, the following code starts a separate thread to publish the generated data to the input Pub/Sub topic.
def publish_data(publisher, project_id: str, topic: str, data: Iterable[Any], delay=0.01, enable_message_ordering=False) -> None:
topic_path = publisher.topic_path(project_id, topic)
for i in range(len(data)):
kv = {"f1": data.iloc[i]}
kv["id"] = i # add event id
msg = json.dumps(kv).encode('utf-8')
if enable_message_ordering:
# set ordering key to a fixed string so messages with the same ordering key will be published in order
publisher.publish(topic_path, data=msg, ordering_key="my-order-key")
else:
publisher.publish(topic_path, data=msg)
time.sleep(delay)
publisher_thread = threading.Thread(
target=publish_data,
args=(input_publisher, PROJECT_ID, INPUT_TOPIC, df, 0.001, True),
)
publisher_thread.start()
print(f"Started to publish data to {INPUT_TOPIC}")
Started to publish data to anomaly-input-9625
Launching the Beam Pipeline
This pipeline adapts the core anomaly detection logic from the previous batch example for a real-time, streaming application. The key modification is in the I/O: instead of operating on a static collection, this pipeline reads its input stream from a Pub/Sub topic and writes the results to a separate output topic.
Notice that the pipeline is run on a separate thread so later steps are not blocked.
def message_to_beam_row(msg: bytes) -> beam.Row:
try:
r = beam.Row(**json.loads(msg.decode('utf-8')))
except Exception as e:
print("Wrong msg: %s" % msg)
print(e)
return r
def keyed_result_to_message(t: Tuple[Any, AnomalyResult]) -> bytes:
idx = t[1].example.id
value = t[1].example.f1
label = next(iter(t[1].predictions)).label
return json.dumps({"id":idx, "value": value, "label": label}).encode('utf-8')
def run_beam_pipeline():
input_sub = pubsub_v1.SubscriberClient.subscription_path(PROJECT_ID, INPUT_SUB)
output_topic = pubsub_v1.PublisherClient.topic_path(PROJECT_ID, OUTPUT_TOPIC)
options = PipelineOptions(["--streaming"])
with beam.Pipeline(options=options) as p:
_ = (p | beam.io.ReadFromPubSub(subscription=input_sub)
| "Convert Pubsub Messages to Rows" >> beam.Map(message_to_beam_row).with_output_types(beam.Row)
| beam.WithKeys(0)
| AnomalyDetection(ZScore(features=["f1"],
sub_stat_tracker=IncSlidingMeanTracker(100),
stdev_tracker=IncSlidingStdevTracker(100)))
| "Convert output to Pubsub Messages" >> beam.Map(keyed_result_to_message)
| beam.io.WriteToPubSub(topic=output_topic, with_attributes=False)
)
pipeline_thread = threading.Thread(
target=run_beam_pipeline,
)
pipeline_thread.start()
print(f"Started to run beam pipeline for anomaly detection")
Started to run beam pipeline for anomaly detection
Collecting Results and Plotting
To prepare for visualization, start another thread that retrieves output from the output pubsub topic.
x = []
y = []
c = []
def collect_result(subscriber):
subscription_path = pubsub_v1.SubscriberClient.subscription_path(PROJECT_ID, OUTPUT_SUB)
NUM_MESSAGES = 100
while True:
response = subscriber.pull(
request={"subscription": subscription_path, "max_messages": NUM_MESSAGES},
retry=retry.Retry(deadline=300),
)
ack_ids = []
for received_message in response.received_messages:
ack_ids.append(received_message.ack_id)
msg = json.loads(received_message.message.data.decode('utf-8'))
x.append(msg['id'])
y.append(msg['value'])
c.append('red' if msg['label'] == 1 else 'green')
if len(ack_ids) > 0:
# Acknowledges the received messages so they will not be sent again.
subscriber.acknowledge(
request={"subscription": subscription_path, "ack_ids": ack_ids}
)
result_thread = threading.Thread(
target=collect_result,
args=(output_subscriber,),
)
result_thread.start()
Run the following line to check how many output results are coming out from the output pubsub.
print(len(x))
168
This following code visualizes the streaming output by repeatedly generating an animation. It refreshes the visualization every 20 seconds to incorporate newly arrived results. Within each refresh, an new animated scatter plot is rendered, progressively drawing each data point to show the evolution of the stream.In these plots, outliers are highlighted in red.
# This will generate a plot every 20 seconds to show how the data stream is processed.
for i in range (5):
matplotlib.rcParams['animation.embed_limit'] = 300
data = np.array(list(zip(x,y)))
fig, ax = plt.subplots()
fig.set_size_inches(12, 4)
ax.axis([0,1000,-10,20])
l = ax.scatter([],[])
l.set_sizes([3])
def animate(i):
i = i * 10
l.set_offsets(data[:i+1])
l.set_color(c)
plt.close() # to avoid extra frame after animation
ani = matplotlib.animation.FuncAnimation(fig, animate, frames=int(len(x)/10), interval=50, repeat=False)
display(HTML(ani.to_jshtml()))
time.sleep(20)
After all the data is processed, run the code below to draw the final scatterplot.
plt.figure(figsize=(12, 4))
plt.xlim(0, 1000)
plt.ylim(-10, 20)
plt.scatter(x=x, y=y, c=c, s=3)
<matplotlib.collections.PathCollection at 0x7ef2eca5f1d0>
Cleaning Up Pub/Sub Resources
# deleting input and output subscriptions
subscriber = pubsub_v1.SubscriberClient()
with subscriber:
try:
subscription_path = subscriber.subscription_path(PROJECT_ID, INPUT_SUB)
subscriber.delete_subscription(request={"subscription": subscription_path})
print(f"Input subscription deleted: {subscription_path}.")
except NotFound:
pass
try:
subscription_path = subscriber.subscription_path(PROJECT_ID, OUTPUT_SUB)
subscriber.delete_subscription(request={"subscription": subscription_path})
print(f"Output subscription deleted: {subscription_path}.")
except NotFound:
pass
Input subscription deleted: projects/apache-beam-testing/subscriptions/anomaly-input-9625-sub. Output subscription deleted: projects/apache-beam-testing/subscriptions/anomaly-output-9625-sub.
# deleting input and output topics
publisher = pubsub_v1.PublisherClient()
with publisher:
try:
topic_path = publisher.topic_path(PROJECT_ID, INPUT_TOPIC)
publisher.delete_topic(request={"topic": topic_path})
print(f"Input topic deleted: {topic_path}")
except NotFound:
pass
try:
topic_path = publisher.topic_path(PROJECT_ID, OUTPUT_TOPIC)
publisher.delete_topic(request={"topic": topic_path})
print(f"Output topic deleted: {topic_path}")
except NotFound:
pass
Input topic deleted: projects/apache-beam-testing/topics/anomaly-input-9625 Output topic deleted: projects/apache-beam-testing/topics/anomaly-output-9625