Detecting Anomalies in Media Streams With Apache Spark

Media companies often process millions of user sessions per day. Each session can reveal valuable insights: how often are requests made? What formats are being streamed? Is someone downloading too much data too quickly, which could indicate stream piracy?

These patterns, when analyzed at scale, can help detect suspicious behavior, optimize CDN delivery, and identify content performance issues. Considering that streaming piracy alone is projected to cost businesses more than $100 billion in revenue per year by 2027, and that poor streaming performance leads to higher customer churn, solving these issues is critical for the bottom line.

At Hydrolix, many of our streaming media customers want to better understand user behavior across sessions—not only to improve user experience, but also to detect anomalies like stream piracy or bot activity. This blog post will walk through a real-world project that uses the Hydrolix Connector for Apache Spark to extract raw data, build session profiles, and cluster similar behaviors using K-means clustering, an unsupervised machine learning algorithm.

The tools used here include Microsoft Fabric, Hydrolix, and the Hydrolix Connector for Apache Spark, which allows you to read Hydrolix data from tools compatible with Apache Spark such as Microsoft Fabric and Databricks. It will also soon be compatible with Amazon EMR and SageMaker.

Hydrolix is a real-time data platform designed for high volumes of log data, making it a good fit when it comes to providing large datasets for machine learning, data science, and more. Visualizations in this post use Grafana, though it’s also possible to visualize Hydrolix data with other tools such as Superset and Kibana.

The Anomaly Detection Process: An Overview

To detect anomalies, this project uses a data pipeline that:

1. Extracts raw session data from Hydrolix using the Hydrolix Connector for Apache Spark.
2. Uses feature engineering to turn requests into structured events per session.
3. Applies Word2Vec to vectorize sessions.
4. Uses K-means clustering to group sessions based on behavioral similarity.
5. Estimates the optimal number of clusters using silhouette and elbow techniques.
6. Visualizes clusters to spot outliers.
7. Detects anomalies such as bots and stream piracy.
8. Stores processed data in Hydrolix for further analysis, mitigation, and detection.

This approach allows for dynamic user behavior profiling without requiring labeled data.

You can find all the code for this example notebook in GitHub. If you want to follow along with the code using Hydrolix, you’ll need to set up the Hydrolix Connector for Apache Spark with Microsoft Fabric first.

Extracting Raw Session Data

The first step is to determine which data should be extracted from raw logs stored in Hydrolix. This example will focus on attributes that will help profile user sessions and add custom labels in later stages.

Here are the attributes:

Using Feature Engineering to Create Structured Events per Session

In order to prepare the data for K-means clustering, the next step is data transformation and feature engineering.

This example uses three token prefixes: TD_ (time difference), EXT_ (file extension), and KB_ (kilobyte size) which will describe each request for a user session as an event.

Note: Rounding reduces the vocabulary size and reduces the computation resources needed for tokenization in later steps.

Here is the code for this part of the pipeline:

window_spec = Window.partitionBy("session").orderBy("timestamp")
df = df.withColumn("prev_timestamp", F.lag("timestamp").over(window_spec))
# Time difference in milliseconds to the nearest 100 milliseconds.
df = df.withColumn("time_diff_ms", F.when(F.col("prev_timestamp").isNull(), F.lit(0))
   .otherwise(F.round((((F.col("timestamp").cast("long") - F.col("prev_timestamp").cast("long")) * 1000) / 100), 0) * 100))
# Convert bytes_out into kilobytes rounded to the nearest hundred.
df = df.withColumn("kb",
                   F.round((F.col("bytes_out") / 1024) / 100, 0) * 100
                  )
                  
df = df.withColumn("event_tokens",
                   F.array(
                       F.concat(F.lit("TD_"), F.col("time_diff_ms")),
                       F.concat(F.lit("EXT_"), F.col("extension")),
                       F.concat(F.lit("KB_"), F.col("kb"))
                   )
                  )
# Create a struct column that holds the timestamp and event_tokens.
df = df.withColumn("ts_token", F.struct(F.col("timestamp"), F.col("event_tokens")))

The next code snippet arranges events in chronological order per user session and creates a new dataframe.

# Group by session and collect the list of ts_token structs
df_grouped = df.groupBy("session").agg(F.collect_list("ts_token").alias("events"))
# Sort the events by timestamp, extract the event_tokens field from each, and flatten into a single array of strings
df_sessions = df_grouped.withColumn(
    "tokens",
    F.flatten(F.expr("transform(array_sort(events), x -> x.event_tokens)"))
).select("session", "tokens")
# Add the session length (number of tokens).
df_sessions = df_sessions.withColumn("session_length", F.size("tokens"))
# Also, compute a representative ua_category for each session (e.g., the first one).
df_ua = df.groupBy("session").agg(F.first("ua_category").alias("ua_category"), 
                                  F.first("client_ip").alias("client_ip"),
                                  F.first("cmcd_sf").alias("cmcd_sf")
                                )
df_sessions = df_sessions.join(df_ua, on="session", how="left")

The result of this code is a dataframe with an array of tokens that is ready for machine learning.

root
 |-- session: string (nullable = true)
 |-- tokens: array (nullable = false)
 |    |-- element: string (containsNull = true)
 |-- session_length: integer (nullable = false)
 |-- ua_category: string (nullable = true)
 |-- client_ip: string (nullable = true)
 |-- cmcd_sf: string (nullable = true)

Apply Word2Vec to Vectorize Sessions

Next, the array of events (tokens) will be transformed into vectors using the Word2Vec model. This prepares the data for clustering and will make it easier to find similarities between clusters.

word2vec = Word2Vec(vectorSize=100, minCount=1, numPartitions=40, inputCol="tokens", outputCol="session_vec")
w2v_model = word2vec.fit(df_sessions)
df_sessions = w2v_model.transform(df_sessions)

The above code adds a new column session_vec to the dataframe that can be used for K-means clustering and other anomaly detection techniques in future steps. 

Estimating the Optimal Number of Clusters

Now it’s time to determine the optimal number of clusters. By default, the number of clusters can be large, which can make categorization potentially less useful. The elbow and silhouette methods are useful for determining the optimal number of clusters—the “K” in K-means clustering.

The elbow method plots the within-cluster sum of squares (WSS) or distortion against the number of clusters (K). The “elbow” point on the curve is where adding more clusters yields diminishing returns, suggesting the optimal K.

Silhouette analysis measures the average silhouette score, which quantifies how well each data point is clustered by considering both the cohesion within its own cluster and the separation from other clusters. A score near +1 indicates that the data point is well-clustered, while a score near 0 indicates overlapping clusters. Finally, a negative score suggests the data point might be in the wrong cluster. 

Together, these methods are complementary when it comes to determining the optimal number of clusters.

The following code uses both approaches to visualize the optimal cluster size.

# Define a range for the number of clusters to try.
cluster_range = list(range(2, 20))  # Trying k from 2 to 10
sse_list = []            # For the Elbow method (sum of squared errors)
silhouette_scores = []   # For the Silhouette method
for k in cluster_range:
    # Configure KMeans with k clusters using feature column.
    kmeans = KMeans(featuresCol="session_vec", k=k, seed=42)
    model = kmeans.fit(df_sessions)
    
    # Get predictions for the current model.
    predictions = model.transform(df_sessions)
    
    # Sum of Squared Errors (SSE) from the training summary.
    sse = model.summary.trainingCost
    sse_list.append(sse)
    
    # Evaluate the Silhouette score (using squaredEuclidean distance).
    evaluator = ClusteringEvaluator(featuresCol="session_vec", metricName="silhouette", distanceMeasure="squaredEuclidean")
    silhouette = evaluator.evaluate(predictions)
    silhouette_scores.append(silhouette)
    
    print(f"Clusters: {k}, SSE: {sse:.2f}, Silhouette Score: {silhouette:.4f}")
# Plot the Elbow method and Silhouette scores.
plt.figure(figsize=(12, 5))
# Elbow curve: k vs. SSE
plt.subplot(1, 2, 1)
plt.plot(cluster_range, sse_list, marker='o')
plt.xlabel("Number of Clusters (k)")
plt.ylabel("Sum of Squared Errors (SSE)")
plt.title("Elbow Method for Optimal k")
# Silhouette score: k vs. Silhouette
plt.subplot(1, 2, 2)
plt.plot(cluster_range, silhouette_scores, marker='o', color='green')
plt.xlabel("Number of Clusters (k)")
plt.ylabel("Silhouette Score")
plt.title("Silhouette Method for Optimal k")
plt.tight_layout()
plt.show()

Here are the visualizations. The diagram on the left shows the optimal K using the elbow method, while the one on the right shows the optimal K using the silhouette method.

Based on these visualizations, the optimal number of clusters is around 6 where the elbow is just about to form and there’s an acceptable silhouette score of 0.6938.

Visualizing Clusters and Outliers

Outliers can be particularly interesting because they may represent anomalies. Now that you know the optimal number of clusters, you can visualize the data on a scatter plot using principal component analysis (PCA). You can do this using vectorized session events (session_vec) and reducing the dimensionality to two for an interactive two-dimensional scatterplot, which is shown in the next image.

With PCA, you can visualize the outliers on the chart, then capture details about these specific outliers by making additional requests to the underlying log data to get more information.

The original dataframe created discussed earlier in this post also includes ua_category, so you can visualize the same data by color-coding it according to this category. This will make it easier to spot user_agent data that’s anomalous and possibly indicative of fraudulent behavior. The next image shows sessions clustered by user agent.

This visualization shows desktop_browser and set_top_box clustered in a similar space, while mobile_iphone and tablet_ios have different behavior characteristics. Grouping data in this way has the potential to give us new and interesting insights such as out of place user agents, or visually identify clusters that have atypical behavior.

Detecting Anomalies Like Bots and Stream Piracy

In addition to visually identifying data points, you can use techniques such as GaussianMixture, which allows us to estimate the probability of a datapoint belonging to a specific cluster. By defining a threshold, you can automatically detect sessions that could be outliers or anomalies. One such anomaly type is stream piracy, which may include requests for uncommon resources, requests for many bytes, and atypical user agents—or some combination of attributes that aren’t necessarily obvious without machine learning.

Here’s the code for configuring GaussianMixture and defining a threshold.

# Define feature column.
feature_column = 'session_vec'
# Configure GaussianMixture. Optionally, tune max iterations if needed.
gmm = GaussianMixture(featuresCol=feature_column, 
                      k=k, 
                      seed=42,
                      maxIter=50)  # Adjust maxIter based on your data
# Fit the model.
anomaly_model = gmm.fit(df_sessions)
# Transform the DataFrame to add a column for anomaly scores.
df_sessions = anomaly_model.transform(df_sessions).cache()
# Materialize the cache to ensure subsequent actions are fast.
df_sessions.count()  
from pyspark.ml.functions import vector_to_array
df_sessions = df_sessions.withColumn("max_prob", F.array_max(vector_to_array(F.col("probability"))))
# Define a threshold, e.g., the 1st percentile of max_prob (you may want to adjust this).
# Note: In Spark, we compute quantiles using approxQuantile.
quantiles = df_sessions.approxQuantile("max_prob", [0.01], 0.001)  # [0.01] is the quantile to compute (1st percentile), 0.001 is the relative error.
print("Quantiles:", quantiles)
threshold = quantiles[0] if quantiles else 0.0
# Mark anomalies: if max_prob is below threshold, label as anomaly (1), else normal (0).
df_sessions = df_sessions.withColumn("anomaly_gmm", F.when(F.col("max_prob") < threshold, 1).otherwise(0))
# Count the number of anomalies detected.
anomaly_count = df_sessions.filter(F.col("anomaly_gmm") == 1).count()
print(f"Number of anomalies detected: {anomaly_count}")

The next image shows a visualization of anomalies using Gaussian mixture.

Storing Processed Data in Hydrolix

After detecting anomalies and possibly creating a batch job to run these algorithms on a regular basis, you can write the data back to Hydrolix for further analysis and action.

Processed data can be enriched and combined with raw data already located in Hydrolix to create a holistic overview of customer behavior, detect and understand anomalies, and make technical and business decisions.

Below is an example dashboard based on the data that could help achieve those goals.

The dashboard shows session clusters, anomaly sessions, session clusters grouped by user agent category, and high-level metrics on anomalies such as the total number of anomalies and the overall anomaly percentage. You can use this dashboard to determine sessions and user agents that have abnormal behaviors, or even just use it to get a quick overview of the total number of anomalies that have been detected.

The next diagram shows which client IPs have the most sessions—a potential indicator of an issue like stream piracy. It would be possible to either manually block IPs or set up an automation to do so.

The next visualization shows an example of using Grafana interactive capabilities to drill down into individual sessions for more details on the raw logs.

You could then query the raw Hydrolix log data by useful attributes such as the client IP. The next image shows individual session logs for a specific client IP during the queried time window.

Because Hydrolix stores log data long-term and cost-effectively, it’s possible to query sessions over a long period of time—fifteen months by default (or longer if needed). A Hydrolix user could discover an anomalous log session that’s an indicator of a low-and-slow attack, account leaching or piracy, and then review all the sessions of that IP over an extended period of time to better understand the possible impact and what mitigation steps are necessary (such as blocking the IP, suspending an account, and so on).

Using Hydrolix for Machine Learning and Anomaly Detection

By using the Hydrolix Connector for Apache Spark to combine the advantages of Hydrolix and Apache Spark tools, you can analyze your logs using machine learning to uncover new insights, detect anomalies such as stream piracy and other fraudulent behavior, answer technical and business challenges, and much more. Hydrolix makes it possible to keep petabytes of data “hot” long-term and cost-effectively, which makes it a good fit for machine learning use cases.

• Learn about using Hydrolix for AIOps.
• Learn more about using the Hydrolix Connector for Apache Spark with Databricks, including how the connector works and how Hydrolix performs against Delta Lake.
• Read about using the Hydrolix Connector for Apache Spark with Microsoft Fabric.
• Check out the Hydrolix Connector for Apache Spark documentation.
• Contact us about a managed trial or demo to see Hydrolix in action.