Getting Started with Apache Spark

Use case provided by: Dan Mallinger, Director of Data Science at Think Big, a Teradata company

Unsupervised Anomaly Detection with Spark

The term anomalous data refers to data that are different from what are expected or normally occur. Detecting anomalies is important in most industries. For example, in network security, anomalous packets or requests can be flagged as errors or potential attacks. In customer security, anomalous online behavior can be used to identify fraud. And in manufacturing and the Internet of Things, anomaly detection is useful for identifying machine failures.

Anomaly detection flags "bad" data—whether that refers to data quality, malicious action, or simply user error. And because bad data are typically—and hopefully—rare, modeling bad data can be difficult when data are sampled. But the act of sampling eliminates too many or all of the anomalies needed to build a detection engine. Instead, you want large data sets—with all their data quality issues—on an analytics platform that can efficiently run detection algorithms. Apache Spark, as a parallelized big data tool, is a perfect match for the task of anomaly detection.

By framing anomalies as "bad data," it becomes clear that the patterns of what we call "bad data" change over time. It’s an arms race: the act of fixing an error will likely result in new errors and stopping one type of attack may lead malicious actors to devise new attacks. Because of an ever-changing environment, detection should not be fixated on one anomalous pattern. What is required then is an unsupervised approach (i.e., one that does not require training data with records flagged as anomalous or not).

Getting Started

Using the KDD'99 data set, we will filter out a number of columns for two reasons: ease of example and removal of labeled data, as this use case only demonstrates unsupervised approaches.

import numpy as np
INPUT = "hdfs://localhost/data/kdd"

def parse_line(line):
    bits = line.split(",")
    return np.array([float(e) for e in bits[4:12]])

df = sc.textFile(INPUT).map(parse_line)

Identifying Univariate Outliers

To begin, we want to get summary statistics for columns in our data. The stats() function provides these summaries. We can then use the results to get all records where the first column lies more than two standard deviations from the mean:

stats = df.map(lambda e: e[0]).stats()
mean, stdev = stats.mean(), stats.stdev()
outliers = df.filter(lambda e: not (mean - 2 * stdev > e[0] > mean + 2 * stdev))
outliers.collect()

Unfortunately, this approach has two limitations. First, it requires knowledge of the parameter distribution: in this case, it assumes data follow a roughly normal distribution. Second, in this example we can only determine outliers by looking at individual variables, but anomalous data is often defined by the relationship between variables. For example, going to the money transfer page of a bank is normal, but doing so without first visiting an account page can signal a fraudulent agent. Thus, we should move ahead with a multivariate approach instead.

Detection with Clustering

Clustering refers to an unsupervised approach whereby data points close to one another are grouped together. A common approach called k-means will produce k clusters where the distance between points is calculated using Euclidean distance. This approach can be quickly run in Spark via MLlib:

from pyspark.mllib.clustering import KMeans
clusters = KMeans.train(df, 5, maxIterations=10,
    runs=1, initializationMode="random")

There are now five clusters created. We can see the size of each cluster by labeling each point and counting. However, note that determining the optimal number of clusters requires iteratively building a model, omitted here because we move to an alternative approach in the next section.

cluster_sizes = df.map(lambda e: clusters.predict(e)).countByValue()

K-means is sensitive to outliers. This means that anomalous data often ends up in clusters alone. Looking at the cluster_sizes dictionary, we find clusters 0 and 2 have 1 and 23 data points each, suggesting they are anomalous. But we cannot be certain without further inspection.

However, because we have a small number of clusters (and probably too few), we may find anomalous data within non-anomalous clusters. A first inspection should look at the distribution of distances of data points from cluster centers:

def get_distance(clusters):
    def get_distance_map(record):
        cluster = clusters.predict(record)
        centroid = clusters.centers[cluster]
        dist = np.linalg.norm(record - centroid)
        return (dist, record)
    return get_distance_map

data_distance = df.map(get_distance(clusters))
hist = data_distance.keys().histogram(10)

Looking at the histogram (represented as a Python list) shows that occurrences of values decline as the distance grows, with the exception of a spike in the last bucket. Those data points are a good starting place to seek anomalies.

K-means clustering is a powerful tool. However, it does have a number of challenges. First, selecting the appropriate value for k can be difficult. Second, K-means is sensitive to the scale of variables (see Exercise 2). Last, there are no thresholds or scores that can be readily used to evaluate new data as anomalous. Because of these limitations, let us move on to an alternative approach.

Exercise 1: Run these calculations and explore the data points. Can you explain the bump at the end of the histogram?

Exercise 2: K-means is sensitive to the scale of variables. This means variables with larger variance will overtake your calculations. Try standardizing variables by subtracting the mean and dividing by the standard deviation for each. How does this change your results?

Detection with Unsupervised Random Forests

Random forests are a powerful prediction method that uses collections of trees with a random parameter holdout to build models that often outperform individual decision trees. However, the random forest is normally a supervised approach, requiring labeled data.

In this section, we introduce a method for turning a supervised model into an unsupervised model for anomaly detection. Unsupervised random forests have a number of advantages over k-means for simple detection. First, they are less sensitive to variable scale. Second, they can fit to "normal" behavior and thus can provide a probability of a data point being anomalous.

To create an unsupervised model from a supervised one, we create a new dataset of "dummy" data. We create dummy data by sampling from within columns of the original data and joining those columns. This creates a dataset whose column values are non-anomalous, but whose relationships between columns are. The unisample function provides an example:

def unisample(df, fraction=1.0):
    columns = df.first()
    new_df = None
    for i in range(0, len(columns)):
        column = df.sample(withReplacement=True, fraction=fraction)
            .map(lambda row: row[i])
            .zipWithIndex()
            .map(lambda e: (e[1], [e[0]]))
        if new_df is None:
            new_df = column
        else:
            new_df = new_df.join(column)
            new_df = new_df.map(lambda e: (e[0], e[1][0] + e[1][1]))
    return new_df.map(lambda e: e[1])

Next we label the dummy data 0 (for "not real") and original data 1 (for "real"). The supervised2unsupervised function shows how to create such a dataset. It also takes a supervised model and returns an unsupervised model, which we will do next:

def supervised2unsupervised(model):
    def run(df, *args, **kwargs):
        unisampled_df = unisample(df)
        unisampled_df = unisample(df)
        labeled_data = df.map(lambda e: LabeledPoint(1, e))
            .union(unisampled_df.map(lambda e: LabeledPoint(0, e)))
        return model(labeled_data, *args, **kwargs)
    return run

Now we can create and fit an unsupervised random forest using the RandomForest function provided by MLlib:

from pyspark.mllib.regression import LabeledPoint
from pyspark.mllib.tree import RandomForest

unsupervised_forest = supervised2unsupervised(RandomForest.trainClassifier)
rf_model = unsupervised_forest(df, numClasses=2, categoricalFeaturesInfo={},
                  numTrees=10, featureSubsetStrategy="auto",
                  impurity='gini', maxDepth=15, maxBins=50)

We have now created a new model (rf_model) that is trained to recognize "normal" data, giving us a robust anomaly detection tool without requiring labeled data. And anytime we run rf_model.predict, we can get a probability of a data point being anomalous!

Exercise 3: The forest in this example has suboptimal parameters (e.g., only have 10 trees). Reconfigure the forest to improve the overall prediction accuracy of non-anomalous data.

Exercise 4: This approach is sensitive to the dummy data generated. Either by "unisampling" with subsets or other approaches, try and bootstrap many forests. Create a single predict function that returns the mean and standard deviation across this collection of models. Use this to generate confidence intervals over your predictions.