open sourcing in February marked a significant step for those looking for accessible yet potent libraries. Rerun's Python visualization Why is visualization important? Visualization is essential since companies like Scale.ai, Weights & Biases, and Hugging Face have streamlined deep learning by addressing dataset labeling, experiment tracking, and pre-trained models. However, a void still exists in rapid data capture and visualization. Many companies develop in-house data visualization solutions but often end up with suboptimal tools due to high development costs. Moreover, Python visualization on data is a problem that is not solved well either, leading to t-based solutions in notebooks. Rerun leverages a Python interface into a high-performant visualization engine (much like Bytewax!) that makes it dead easy to analyze streaming data. streaming JavaScrip Rust In this blog post, we will explore how to use Bytewax and Rerun to visualize real-time streaming data in Python and create a real-time anomaly detection visualization. We chose anomaly detection, a.k.a. outlier detection because it is a critical component in numerous applications, such as cybersecurity, fraud detection, and monitoring of industrial processes. Visualizing these anomalies in real-time can aid in quickly identifying potential issues and taking necessary actions to mitigate them. For those eager to dive in, check out our end-to-end Python solution on our . Don't forget to star Bytewax! GitHub Overview Here is what we'll cover: We will navigate the code and briefly discuss top-level entities Then we will discuss each step of the dataflow in greater detail: initialization of our dataflow, input source, stateful anomaly detection, data visualization & output, and how to spawn a cluster Finally, we will learn how to run it and see the beautiful visualization, all in Python <3 As a bonus, we will think about other use cases Let's go! Setup your environment This blog post is based on the following versions of Bytewax and Rerun: bytewax==0.15.1
rerun-sdk==0.4.0 Rerun and Bytewax are installable as pip install rerun-sdk
pip install bytewax Follow Bytewax for updates as we are baking a new version that will ease the development of data streaming apps in Python further. Code The solution is relatively compact, so we copy the entire code example here. Please feel free to skip this big chunk if it looks overwhelming; we will discuss each function later. import random
# pip install rerun-sdk
import rerun as rr

from time import sleep
from datetime import datetime

from bytewax.dataflow import Dataflow
from bytewax.execution import spawn_cluster
from bytewax.inputs import ManualInputConfig, distribute
from bytewax.outputs import ManualOutputConfig


rr.init("metrics")
rr.spawn()

start = datetime.now()

def generate_random_metrics(worker_index, worker_count, resume_state):
    assert resume_state is None
    keys = ["1", "2", "3", "4", "5", "6"]
    this_workers_keys = distribute(keys, worker_index, worker_count)

    for _ in range(1000):
        for key in this_workers_keys:
            value = random.randrange(0, 10)
            if random.random() > 0.9:
                value *= 2.0
            yield None, (key, (key, value, (datetime.now() - start).total_seconds()))
            sleep(random.random() / 10.0)


class ZTestDetector:
    """Anomaly detector.

    Use with a call to flow.stateful_map().

    Looks at how many standard deviations the current item is away
    from the mean (Z-score) of the last 10 items. Mark as anomalous if
    over the threshold specified.
    """

    def __init__(self, threshold_z):
        self.threshold_z = threshold_z

        self.last_10 = []
        self.mu = None
        self.sigma = None

    def _push(self, value):
        self.last_10.insert(0, value)
        del self.last_10[10:]

    def _recalc_stats(self):
        last_len = len(self.last_10)
        self.mu = sum(self.last_10) / last_len
        sigma_sq = sum((value - self.mu) ** 2 for value in self.last_10) / last_len
        self.sigma = sigma_sq**0.5

    def push(self, key__value__t):
        key, value, t = key__value__t
        is_anomalous = False
        if self.mu and self.sigma:
            is_anomalous = abs(value - self.mu) / self.sigma > self.threshold_z

        self._push(value)
        self._recalc_stats()
        rr.log_scalar(f"temp_{key}/data", value, color=[155, 155, 155])
        if is_anomalous:
            rr.log_point(f"3dpoint/anomaly/{key}", [t, value, float(key) * 10], radius=0.3, color=[255,100,100])
            rr.log_scalar(
                f"temp_{key}/data/anomaly",
                value,
                scattered=True,
                radius=3.0,
                color=[255, 100, 100],
            )
        else:
            rr.log_point(f"3dpoint/data/{key}", [t, value, float(key) * 10], radius=0.1)

        return self, (value, self.mu, self.sigma, is_anomalous)


def output_builder(worker_index, worker_count):
    def inspector(input):
        metric, (value, mu, sigma, is_anomalous) = input
        print(
            f"{metric}: "
            f"value = {value}, "
            f"mu = {mu:.2f}, "
            f"sigma = {sigma:.2f}, "
            f"{is_anomalous}"
        )

    return inspector


if __name__ == '__main__':
    flow = Dataflow()
    flow.input("input", ManualInputConfig(generate_random_metrics))
    # ("metric", value)
    flow.stateful_map("AnomalyDetector", lambda: ZTestDetector(2.0), ZTestDetector.push)
    # ("metric", (value, mu, sigma, is_anomalous))
    flow.capture(ManualOutputConfig(output_builder))
    spawn_cluster(flow) The provided code demonstrates how to create a real-time anomaly detection pipeline using Bytewax and Rerun. Let's break down the essential components of this code: : This function generates random metrics simulating real-world data streams. It generates data points with a small chance of having an anomaly (values doubled). generate_random_metrics : This class implements an anomaly detector using the Z-score method. It maintains the mean and standard deviation of the last 10 values and marks a value as anomalous if its Z-score is greater than a specified threshold. ZTestDetector : This function is used to define the output behavior for the data pipeline. In this case, it prints the metric name, value, mean, standard deviation, and whether the value is anomalous. output_builder : The main part of the code constructs the dataflow using Bytewax, connecting the RandomMetricInput, ZTestDetector, and the output builder. Dataflow : The Rerun visualization is integrated into the ZTestDetector class. The rr.log_scalar and rr.log_point functions are used to plot the data points and their corresponding anomaly status. Rerun visualization Now, with an understanding of the code's main components, let's discuss how the visualization is created step by step. Building the Dataflow To create a dataflow pipeline, you need to: Initialize a new dataflow with . flow = Dataflow() Define the input source using . flow.input("input", ManualInputConfig(generate_random_metrics)) Apply the stateful anomaly detector using . flow.stateful_map("AnomalyDetector", lambda: ZTestDetector(2.0), ZTestDetector.push) Configure the output behavior with . flow.capture(ManualOutputConfig(output_builder)) Finally, spawn a cluster to execute the dataflow with . spawn_cluster(flow, proc_count=3) The resulting dataflow reads the randomly generated metric values from , passes them through the for anomaly detection, and outputs the results using the function. Let's clarify the details for each step. input_builder ZTestDetector output_builder function generate_random_metrics The function serves as an alternative input source for the dataflow pipeline, generating random metric values in a distributed manner across multiple workers. It accepts three parameters: , , and . generate_random_metrics worker_index worker_count resume_state def generate_random_metrics(worker_index, worker_count, resume_state):
    assert resume_state is None
    keys = ["1", "2", "3", "4", "5", "6"]
    this_workers_keys = distribute(keys, worker_index, worker_count)

    for _ in range(1000):
        for key in keys:
            value = random.randrange(0, 10)
            if random.random() > 0.9:
                value *= 2.0
            yield None, (key, (key, value, (datetime.now() - start).total_seconds()))
            sleep(random.random() / 10.0) : The index of the current worker in the dataflow pipeline. worker_index : The total number of workers in the dataflow pipeline. worker_count : The state of the input source from which to resume. In this case, it is asserted to be , indicating that the input source does not support resuming from a previous state. resume_state None Here's a step-by-step description of the generate_random_metrics function: Assert that is . resume_state None Define a list of keys representing the metrics. Distribute the keys among the workers using the distribute function (not provided in the code snippet). The distributed keys for the current worker are assigned to this_workers_keys. Iterate 1,000 times and, for each iteration, iterate through the list of keys: Generate a random value between 0 and 10. With a 10% probability, double the value to simulate an anomaly. Yield a tuple containing None (to indicate no specific partition key), the key, the generated value, and the elapsed time since the starting time (not provided in the code snippet). Introduce a sleep time between each generated value to simulate real-time data generation. The function is used in the dataflow as the input source with the following line of code: generate_random_metrics flow.input("input", ManualInputConfig(generate_random_metrics)) This line tells the dataflow to use the class to generate the input data for the pipeline. RandomMetricInput Class ZTestDetector The class is an anomaly detector that uses the Z-score method to identify whether a data point is anomalous or not. The Z-score is the number of standard deviations a data point is from the mean of a dataset. If a data point's Z-score is higher than a specified threshold, it is considered anomalous. ZTestDetector The class has the following methods: : The constructor initializes the ZTestDetector with a threshold Z-score value. It also initializes the last 10 values list (self.last_10), mean (self.mu), and standard deviation (self.sigma). __init__(self, threshold_z) : This private method is used to update the list of last 10 values with the new value. It inserts the new value at the beginning of the list and removes the oldest value, maintaining the list length at 10. _push(self, value) : This private method recalculates the mean and standard deviation based on the current values in the self.last_10 list. _recalc_stats(self) : This public method takes a tuple containing a key, a value, and a timestamp as input. It calculates the Z-score for the value, updates the last 10 values list, and recalculates the mean and standard deviation. It also logs the data point and its anomaly status using Rerun's visualization functions. Finally, it returns the updated instance of the ZTestDetector class and a tuple containing the value, mean, standard deviation, and anomaly status. push(self, key__value__t) The ZTestDetector class is used in the dataflow pipeline as a stateful map with the following code: flow.stateful_map("AnomalyDetector", lambda: ZTestDetector(2.0), ZTestDetector.push) This line tells the dataflow to apply the with a Z-score threshold of and use the method to process the data points. ZTestDetector 2.0 push Visualizing Anomalies To visualize the anomalies, the class logs the data points and their corresponding anomaly status using Rerun's visualization functions. Specifically, is used to plot a scalar value, while is used to plot 3D points. ZTestDetector rr.log_scalar rr.log_point The following code snippet shows how the visualization is created: rr.log_scalar(f"temp_{key}/data", value, color=[155, 155, 155])
if is_anomalous:
    rr.log_point(f"3dpoint/anomaly/{key}", [t, value, float(key) * 10], radius=0.3, color=[255,100,100])
    rr.log_scalar(
        f"temp_{key}/data/anomaly",
        value,
        scattered=True,
        radius=3.0,
        color=[255, 100, 100],
    )
else:
    rr.log_point(f"3dpoint/data/{key}", [t, value, float(key) * 10], radius=0.1) Here, we first log a scalar value representing the metric. Then, depending on whether the value is anomalous, we log a 3D point with a different radius and color. Anomalous points are logged in red with a larger radius, while non-anomalous points are logged with a smaller radius. Function output_builder The function is used to define the output behavior for the data pipeline. In this specific example, it is responsible for printing the metric name, value, mean, standard deviation, and whether the value is anomalous. output_builder The function takes two arguments: and . These arguments help the function understand the index of the worker and the total number of workers in the dataflow pipeline. worker_index worker_count Here's the definition of the output_builder function: def output_builder(worker_index, worker_count):
    def inspector(input):
        metric, (value, mu, sigma, is_anomalous) = input
        print(
            f"{metric}: "
            f"value = {value}, "
            f"mu = {mu:.2f}, "
            f"sigma = {sigma:.2f}, "
            f"{is_anomalous}"
        )

    return inspector This function is a higher-order function, which means it returns another function called . The function is responsible for processing the input data tuple and printing the desired output. inspector inspector The output builder function is later used in the dataflow pipeline when configuring the output behavior with flow.capture(ManualOutputConfig(output_builder)). Running the Dataflow Bytewax can run as a single process or in a multi-process way. This dataflow has been created to scale across multiple processes, but we will start off running it as a single process with the execution module. spawn_cluster spawn_cluster(flow) If we wanted to increase the parallelism, we would simply add more processes as arguments. For example - . spawn_cluster(flow, proc_count=3) To run the provided code we can simply run it as a Python script, but first we need to install the dependencies. Create a new file in the same directory as dataflow.py and name it requirements.txt. Add the following content to the requirements.txt file: bytewax==0.15.1
rerun-sdk==0.4.0 Open a terminal in the directory containing the requirements.txt and dataflow.py files. Install the dependencies using the following command: pip install -r requirements.txt And run the dataflow! python dataflow.py Expanding the Use Case While the provided code serves as a basic example of real-time anomaly detection, you can expand this pipeline to accommodate more complex scenarios. For example: : Replace the RandomMetricInput class with a custom class that reads data from a real-world source, such as IoT sensors, log files, or streaming APIs. Incorporate real-world data sources : You can replace the ZTestDetector class with other stateful anomaly detection methods, such as moving average, exponential smoothing, or machine learning-based approaches. Implement more sophisticated anomaly detection techniques : Enhance the Rerun visualization by adding more data dimensions, adjusting the color schemes, or modifying the plot styles to better suit your needs. Customize the Visualization : Instead of simply printing the anomaly results, you can integrate the pipeline with alerting or monitoring systems to notify the appropriate stakeholders when an anomaly is detected. Integrate with alerting and monitoring systems By customizing and extending the dataflow pipeline, you can create a powerful real-time anomaly detection and visualization solution tailored to your specific use case. The combination of Bytewax and Rerun offers a versatile and scalable foundation for building real-time data processing and visualization systems. Conclusion This blog post has demonstrated how to use Bytewax and Rerun to create a real-time anomaly detection visualization. By building a dataflow pipeline with Bytewax and integrating Rerun's powerful visualization capabilities, we can monitor and identify anomalies in our data as they occur. Originally written by Zander Matheson here. Join our community: Slack Github

This story contains new, firsthand information uncovered by the writer.

Enjoy Python native data streaming now ->

Too Long; Didn't Read

Download free Tech Leaders Productivity Report!

Visualizing Real-Time Anomaly Detection with Python

About Author

Comments

TOPICS

THIS ARTICLE WAS FEATURED IN

Related Stories

Untitled Story

The Noonification: Anomaly Detection with Privileged Information—Part 1 (12/13/2023)

The Noonification: Craig Wright is Satoshi Nakamoto (12/24/2023)

3 Types of Anomalies in Anomaly Detection

A Zero-Knowledge Anomaly Detection Approach for Robust Federated Learning

Advanced AI Techniques for Financial Fraud Detection

The Noonification: Anomaly Detection with Privileged Information—Part 1 (12/13/2023)

The Noonification: Craig Wright is Satoshi Nakamoto (12/24/2023)

3 Types of Anomalies in Anomaly Detection

A Zero-Knowledge Anomaly Detection Approach for Robust Federated Learning

Advanced AI Techniques for Financial Fraud Detection

Light-Mode

Classic

Newspaper

Dark-Mode

Neon Noir

Minty

HN StartUps