When it comes to anomaly detection, one of the key challenges that many organizations face is that it can be difficult to know how to define what an anomaly is. How do you define and anticipate unusual network intrusions, manufacturing defects, or insurance fraud? If you have labeled data with known anomalies, then you can choose from a variety of supervised machine learning model types that are already supported in BigQuery ML. But what can you do if you don’t know what kind of anomaly to expect, and you don’t have labeled data? Unlike typical predictive techniques that leverage supervised learning, organizations may need to be able to detect anomalies in the absence of labeled data. 

Today we are announcing the public preview of new anomaly detection capabilities in BigQuery ML that leverage unsupervised machine learning to help you detect anomalies without needing labeled data. Depending on whether or not the training data is time series, users can now detect anomalies in training data or on new input data using a new ML.DETECT_ANOMALIES function (documentation), with the following models:

• Autoencoder model, now in Public Preview (documentation)
• K-means model, already GA (documentation)
• ARIMA_PLUS time series model, already GA (documentation)

How does anomaly detection with ML.DETECT_ANOMALIES work?

To detect anomalies in non-time-series data, you can use:

• K-means clustering models: When you use ML.DETECT_ANOMALIES with a k-means model, anomalies are identified based on the value of each input data point’s normalized distance to its nearest cluster. If that distance exceeds a threshold determined by the contamination value provided by the user, the data point is identified as an anomaly. 
• Autoencoder models: When you use ML.DETECT_ANOMALIES with an autoencoder model, anomalies are identified based on the reconstruction error for each data point. If the error exceeds a threshold determined by the contamination value, it is identified as an anomaly. 

To detect anomalies in time-series data, you can use: 

• ARIMA_PLUS time series models: When you use ML.DETECT_ANOMALIES with an ARIMA_PLUS model, anomalies are identified based on the confidence interval for that timestamp. If the probability that the data point at that timestamp occurs outside of the prediction interval exceeds a probability threshold provided by the user, the datapoint is identified as an anomaly.

Below we show code examples of anomaly detection in BigQuery ML for each of the above scenarios.

Anomaly detection with a k-means clustering model

You can now detect anomalies using k-means clustering models, by running ML.DETECT_ANOMALIES to detect anomalies in the training data or in new input data. Begin by creating a k-means clustering model:

Language: SQL

  CREATE MODEL `mydataset.my_kmeans_model`
OPTIONS(
  MODEL_TYPE = 'kmeans',
  NUM_CLUSTERS = 8,
  KMEANS_INIT_METHOD = 'kmeans++'
) AS
SELECT 
  * EXCEPT(Time, Class) 
FROM 
  `bigquery-public-data.ml_datasets.ulb_fraud_detection`;

With the k-means clustering model trained, you can now run ML.DETECT_ANOMALIES to detect anomalies in the training data or in new input data.
To detect anomalies in the training data, use ML.DETECT_ANOMALIES with the same data used during training:

Language: SQL

  SELECT
  *
FROM
  ML.DETECT_ANOMALIES(MODEL `mydataset.my_kmeans_model`,
                      STRUCT(0.02 AS contamination),
                      TABLE `bigquery-public-data.ml_datasets.ulb_fraud_detection`);

To detect anomalies in new data, use ML.DETECT_ANOMALIES and provide new data as input:

Language: SQL

  SELECT
  *
FROM
  ML.DETECT_ANOMALIES(MODEL `mydataset.my_kmeans_model`,
                      STRUCT(0.02 AS contamination),
                      (SELECT * FROM `mydataset.newdata`));

How does anomaly detection work for k-means clustering models? 

Anomalies are identified based on the value of each input data point’s normalized distance to its nearest cluster, which, if exceeds a threshold determined by the contamination value, is identified as an anomaly. How does this work exactly? With a k-means model and data as inputs, ML.DETECT_ANOMALIES first computes the absolute distance for each input data point to all cluster centroids in the model, then normalizes each distance by the respective cluster radius (which is defined as the standard deviation of the absolute distances of all points in this cluster to the centroid). For each data point, ML.DETECT_ANOMALIES returns the nearest centroid_id based on normalized_distance, as seen in the screenshot above. The contamination value, specified by the user, determines the threshold of whether a data point is considered an anomaly. For example, a contamination value of 0.1 means that the top 10% of descending normalized distance from the training data will be used as the cut-off threshold. If the normalized distance for a datapoint exceeds the threshold, then it is identified as an anomaly. Setting an appropriate contamination will be highly dependent on the requirements of the user or business. 

For more information on anomaly detection with k-means clustering, please see the documentation here.

Anomaly detection with an autoencoder model

You can now detect anomalies using autoencoder models, by running ML.DETECT_ANOMALIES to detect anomalies in the training data or in new input data. 

Begin by creating an autoencoder model:

Language: SQL

  CREATE MODEL `mydataset.my_autoencoder_model`
OPTIONS(
  model_type='autoencoder',
  activation_fn='relu',
  batch_size=8,
  dropout=0.2,  
  hidden_units=[32, 16, 4, 16, 32],
  learn_rate=0.001,
  l1_reg_activation=0.0001,
  max_iterations=10,
  optimizer='adam'
) AS 
SELECT 
  * EXCEPT(Time, Class) 
FROM 
  `bigquery-public-data.ml_datasets.ulb_fraud_detection`;

To detect anomalies in the training data, use ML.DETECT_ANOMALIES with  the same data used during training:

Language: SQL

  SELECT
  *
FROM
  ML.DETECT_ANOMALIES(MODEL `mydataset.my_autoencoder_model`,
                      STRUCT(0.02 AS contamination),
                      TABLE `bigquery-public-data.ml_datasets.ulb_fraud_detection`);

To detect anomalies in new data, use ML.DETECT_ANOMALIES and provide new data as input:

Language: SQL

  SELECT
  *
FROM
  ML.DETECT_ANOMALIES(MODEL `mydataset.my_autoencoder_model`,
                      STRUCT(0.02 AS contamination),
                      (SELECT * FROM `mydataset.newdata`));

How does anomaly detection work for autoencoder models? 

Anomalies are identified based on the value of each input data point’s reconstructed error, which, if exceeds a threshold determined by the contamination value, is identified as an anomaly. How does this work exactly? With an autoencoder model and data as inputs, ML.DETECT_ANOMALIES first computes the mean_squared_error for each data point between its original values and its reconstructed values. The contamination value, specified by the user, determines the threshold of whether a data point is considered an anomaly. For example, a contamination value of 0.1 means that the top 10% of descending error from the training data will be used as the cut-off threshold. Setting an appropriate contamination will be highly dependent on the requirements of the user or business. 

For more information on anomaly detection with autoencoder models, please see the documentation here.

Anomaly detection with an ARIMA_PLUS time-series model

With ML.DETECT_ANOMALIES, you can now detect anomalies using ARIMA_PLUS time series models in the (historical) training data or in new input data. Here are some examples of when might you want to detect anomalies with time-series data:

Detecting anomalies in historical data: 

• Cleaning up data for forecasting and modeling purposes, e.g. preprocessing historical time series before using them to train an ML model.  
• When you have a large number of retail demand time series (thousands of products across hundreds of stores or zip codes), you may want to quickly identify which stores and product categories had anomalous sales patterns, and then perform a deeper analysis of why that was the case.

Forward looking anomaly detection: 

• Detecting consumer behavior and pricing anomalies as early as possible: e.g. if traffic to a specific product page suddenly and unexpectedly spikes, it might be because of an error in the pricing process that leads to an unusually low price. 
• When you have a large number of retail demand time series (thousands of products across hundreds of stores or zip codes), you would like to identify which stores and product categories had anomalous sales patterns based on your forecasts, so you can quickly respond to any unexpected spikes or dips.

How do you detect anomalies using ARIMA_PLUS? Begin by creating an ARIMA_PLUS time series model:

Language: SQL

  CREATE OR REPLACE MODEL mydataset.my_arima_plus_model
OPTIONS(
  MODEL_TYPE='ARIMA_PLUS',
  TIME_SERIES_TIMESTAMP_COL='date',
  TIME_SERIES_DATA_COL='total_amount_sold',
  TIME_SERIES_ID_COL='item_name',
  HOLIDAY_REGION='US' 
) AS
SELECT
  date,
  item_description AS item_name,
  SUM(bottles_sold) AS total_amount_sold
FROM
  `bigquery-public-data.iowa_liquor_sales.sales`
GROUP BY
  date,
  item_name
HAVING
  date BETWEEN DATE('2016-01-04') AND DATE('2017-06-01')
  AND item_name IN ("Black Velvet", "Captain Morgan Spiced Rum",
    "Hawkeye Vodka", "Five O'Clock Vodka", "Fireball Cinnamon Whiskey");

To detect anomalies in the training data, use ML.DETECT_ANOMALIES with the model obtained above:

Language: SQL

  SELECT
  *
FROM
  ML.DETECT_ANOMALIES(MODEL `mydataset.my_arima_plus_model`,
                      STRUCT(0.8 AS anomaly_prob_threshold));

To detect anomalies in new data, use ML.DETECT_ANOMALIES and provide new data as input:

Language: SQL

  WITH
  new_data AS (
  SELECT
    date,
    item_description AS item_name,
    SUM(bottles_sold) AS total_amount_sold
  FROM
    `bigquery-public-data.iowa_liquor_sales.sales`
  GROUP BY
    date,
    item_name
  HAVING
    date BETWEEN DATE('2017-06-02')
    AND DATE('2017-10-01')
    AND item_name IN ('Black Velvet',
      'Captain Morgan Spiced Rum',
      'Hawkeye Vodka',
      "Five O'Clock Vodka",
      'Fireball Cinnamon Whiskey') )
SELECT
  *
FROM
  ML.DETECT_ANOMALIES(MODEL `mydataset.my_arima_plus_model`,
    STRUCT(0.8 AS anomaly_prob_threshold),
    (SELECT
      *
    FROM
      new_data));

For more information on anomaly detection with ARIMA_PLUS time series models, please see the documentation here.

Thanks to the BigQuery ML team, especially Abhinav Khushraj, Abhishek Kashyap, Amir Hormati, Jerry Ye, Xi Cheng, Skander Hannachi, Steve Walker, and Stephanie Wang.

We are excited to announce that Google has been named a Leader in The Forrester Wave™: Streaming Analytics, Q2 2021 report. Thank you to our strong community of customers and partners for working with us to deliver a customer focused product. We believe Forrester’s recognition is an acknowledgement of our leadership across an integrated set of capabilities that rely on data to drive transformation. We were also honored to be named a leader in The Forrester Wave™: Cloud Data Warehouse, Q1 2021. 

Forrester gave Dataflow a score of 5 out of 5 across 12 different criteria and according to the report: “Google Cloud Dataflow has strengths in data sequencing, advanced analytics, performance, and high-availability. Google Dataflow’s sweet spot is for enterprises that have a preponderance of real-time data generated on Google Cloud Platform or wish to simplify all data processing by using a single platform that unifies both streaming and batch jobs.” 

Harnessing the power of real-time data

The speed with which businesses are able to respond to change is the difference between those that successfully navigate the future and those that get left behind. In order to accelerate their digital transformation, reimagine their business and leverage the power of real-time data,  today’s data leaders require a streaming analytics platform that provides both depth and breadth.  

Cloud Pub/Sub and Cloud Dataflow, based on more than a decade of experience in internet scale systems for Google’s own needs, provide customers with a reliable, scalable, performant platform. In addition, we’ve designed these products for ease of use to make streaming analytics accessible to more users, which is why customers such as Sky and others from across all industries use Dataflow to run streaming analytics workloads.

5 out of 5 across key streaming analytics criteria

While Forrester gave Dataflow a score of 5 out of 5 in 12 criteria, the product achieved the highest possible scores in areas that are top of mind for our customers.

We continue to be focused on solving problems that matter to you. For example, just in the last month we announced Dataflow Prime and  Auto Sharding for BigQuery – two new auto tuning capabilities that bring efficiency and simplicity to your streaming pipelines. 

Dataflow achieves highest score possible in strategy 

With Google, organizations gain an industry leading product and a partner that has the vision and strategy to help you tackle new business challenges and provide delightful experiences to your customers.

In summary, we are honored to be a Leader in The Forrester Wave™, Streaming Analytics, and look forward to continuing to innovate and partner with you on your digital transformation journey. 

Download the full report: The Forrester Wave™: Streaming Analytics, Q2 2021 and check out these smart analytics reference patterns. To learn more about Dataflow, visit our website and get to know the product by taking an interactive tutorial. You can also watch recordings from the Data Cloud Summit event (May 2021), where we provided an in-depth view of new product innovations in Dataflow and other data analytics products.

As part of the daily business operations on its advertising platform, Twitter serves billions of ad engagement events, each of which potentially affects hundreds of downstream aggregate metrics. To enable its advertisers to measure user engagement and track ad campaign efficiency, Twitter offers a variety of analytics tools, APIs, and dashboards that can aggregate millions of metrics per second in near-real time.

In this post, you’ll get details on how the Twitter Revenue Data Platform engineering team, led by Steve Niemitz, migrated their on-prem architecture to Google Cloud to boost the reliability and accuracy of Twitter’s ad analytics platform.

Deciding to migrate

Over the past decade, Twitter has developed powerful data transformation pipelines to handle the load of its ever-growing user base worldwide. The first deployments for those pipelines were initially all running in Twitter’s own data centers. The input data streamed from various sources into Hadoop Distributed File System (HDFS) as LZO-compressed Thrift files in an Elephant Bird container format. The data was then processed and aggregated in batches by Scalding data transformation pipelines. Then, aggregation results were output into Manhattan, Twitter’s homegrown distributed key-value store, for serving. Additionally, a streaming system using Twitter’s homegrown systems Eventbus (a messaging tool built on top of DistributedLog), Heron (a stream processing engine), and Nighthawk (a sharded Redis deployment) powered the real-time analytics that Twitter had to provide, filling the gap between the current time and the last batch run.

While this system consistently sustained massive scale, its original design and implementation was starting to reach some limits. In particular, some parts of the system that had grown organically over the years were difficult to configure and extend with new features. Some intricate, long-running jobs were also unreliable, leading to sporadic failures. The legacy end-user serving system was very expensive to run and couldn’t support large queries.

To accommodate for the projected growth in user engagement over the next few years and streamline the development of new features, the Twitter Revenue Data Platform engineering team decided to rethink the architecture and deploy a more flexible and scalable system in Google Cloud.

Platform modernization: First iteration

In the middle of 2017, Steve and his team tackled the first redesign iteration of its advertising data platform modernization, leading to Twitter’s collaboration with Google Cloud.

At first, the team left the data aggregation legacy Scalding pipelines unchanged and continued to run them in Twitter’s data centers. But the batch layer’s output was switched from Manhattan to two separate storage locations in Google Cloud:

• BigQuery—Google’s serverless and highly scalable data warehouse, to support ad-hoc and batch queries.
• Cloud Bigtable—Google’s low-latency, fully managed NoSQL database, to serve as a back end for online dashboards and consumer APIs.

The output aggregations from the Scalding pipelines were first transcoded from Hadoop sequence files to Avro on-prem, staged in four-hour batches to Cloud Storage, and then loaded into BigQuery. A simple pipeline deployed on Dataflow, Google Cloud’s fully managed streaming and batch analytics service, then read the data from BigQuery and applied some light transformations. Finally, the Dataflow pipeline wrote the results into Bigtable.

The team built a new query service to fetch aggregated values from Bigtable and process end-user queries. They deployed this query service in a Google Kubernetes Engine (GKE) cluster in the same region as the Bigtable instance to optimize for data access latency.

Here’s a look at the architecture:

This first iteration already brought many important benefits:

• It de-risked the overall migration effort, letting Twitter avoid migrating both the aggregation business logic and storage at the same time.
• The end-user serving system’s performance improved substantially. Thanks to Bigtable’s linear scalability and extremely low latency for data access, the serving system’s P99 latencies decreased from 2+ seconds to 300ms.
• Reliability increased significantly. The team now rarely, if ever, gets paged for the serving system anymore.

Platform modernization: second iteration

With the new serving system in place, in 2019 the Twitter team began to redesign the rest of the data analytics pipeline using Google Cloud technologies. The redesign sought to solve several existing pain points:

• Because the batch and streaming layers ran on different systems, much of the logic was duplicated between systems.
• While the serving system had been moved into the cloud, the existing pain points of the Hadoop aggregation process still existed.
• The real-time layer was expensive to run and required significant operational attention.

With these pain points in mind, the team began evaluating technologies that could help solve them. They considered several open-source stream processing frameworks initially: Apache Flink, Apache Kafka Streams, and Apache Beam. After evaluating all possible options, the team chose Apache Beam for a few key reasons:

• Beam’s built-in support for exactly-once operations at extremely large scale across multiple clusters.
• Deep integration with other Google Cloud products, such as Bigtable, BigQuery, and Pub/Sub, Google Cloud’s fully managed, real-time messaging service.
• Beam’s programming model, which unifies batch and streaming and lets a single job operate on either batch inputs (Cloud Storage), or streaming inputs (Pub/Sub).
• The ability to deploy Beam pipelines on Dataflow’s fully managed service.

The combination of Dataflow’s fully managed approach and Beam’s comprehensive feature set let Twitter simplify the structure of its data transformation pipeline, as well as increase overall data processing capacity and reliability.

Here’s what the architecture looks like after the second iteration:

In this second iteration, the Twitter team re-implemented the batch layer as follows: Data is first staged from on-prem HDFS to Cloud Storage. A batch Dataflow job then regularly loads the data from Cloud Storage, processes the aggregations, and dual-writes the results to BigQuery for ad-hoc analysis and Bigtable for the serving system.

The Twitter team also deployed an entirely new streaming layer in Google Cloud. For data ingestion, an on-prem service now pushes two different streams of Avro-formatted messages to Pub/Sub. Each message contains a bundle of multiple raw events and affects between 100 and 1,000 aggregations. This leads to more than 3 million aggregations per second performed by four Dataflow jobs (J0-3 in the diagram above). All Dataflow jobs share the same topology, although each job consumes messages from different streams or topics.

One stream, which contains critical data, enters the system at a rate of 200,000 messages per second and is partitioned in two separate Pub/Sub topics. A Dataflow job (J3 in the diagram) consumes those two streams, performs 400,000 aggregations per second, and outputs the results to a table in Bigtable.

The other stream, which contains less critical but higher volume data, enters the system at a rate of around 80,000 messages per second and is partitioned into six separate topics. Three Dataflow jobs (J0, J1, and J2) share the processing of this larger stream, with each of them handling two of the available six topics in parallel, then also outputting the results to a table in Bigtable. In total, those three jobs process over 2 million aggregations/second.

Partitioning the high-volume stream into multiple topics offers a number of advantages:

• The partitioning is organized by applying a hash function on the aggregation key and then dividing the function’s result by the number of available partitions (in this case, six). This guarantees that any per-key grouping operation in downstream pipelines is scoped to a single partition, which is required for consistent aggregation results.
• When deploying updates to the Dataflow jobs, admins can drain and relaunch each job individually in sequence, allowing the remaining pipelines to continue uninterrupted and minimizing impact on the end users.
• The three jobs can each handle two topics without issue currently, and there is still room to scale horizontally up to six jobs if needed. The number of topics (six) is arbitrary, but is a good balance at the moment based on current needs and potential spikes in traffic.

To assist with job configuration, Twitter initially considered using Dataflow’s template system, a powerful feature that enables the encapsulation of Dataflow pipelines into repeatable templates that can be configured at runtime. However, since Twitter needed to deploy jobs with topologies that might change over time, the team decided instead to implement a custom declarative system where developers can specify different parameters for their jobs in a pystachio DSL: tuning parameters, data sources to operate on, sink tables for aggregation outputs, and the jobs’ source code location. A new major version of Dataflow templates, called Flex Templates, will remove some of the previous limitations with the template architecture and allow any Dataflow job to be templatized.

For job orchestration, the Twitter team built a custom command line tool that processes the configuration files to call the Dataflow API and submit jobs. The tool also allows developers to submit a job update by automatically performing a multi-step process, like this:

1. Drain the old job:
   • Call the Dataflow API to identify which data sources are used in the job (for example, a Pub/Sub topic reader).
   • Initiate a drain request.
   • Poll the Dataflow API for the watermark of the identified sources until the maximum watermark is hit, which indicates that the draining operation is complete.
2. Launch the new job with the updated code.

This simple, flexible, and powerful system allows developers to focus on their data transformation code without having to be concerned about job orchestration or the underlying infrastructure details.

Looking ahead

Six months after fully transitioning its ad analytics data platform to Google Cloud, Twitter has already seen huge benefits. Twitter’s developers have gained in agility as they can more easily configure existing data pipelines and build new features much faster. The real-time data pipeline has also greatly improved its reliability and accuracy, thanks to Beam’s exactly-once semantics and the increased processing speed and ingestion capacity enabled by Pub/Sub, Dataflow, and Bigtable.

Twitter engineers have enjoyed working with Dataflow and Beam for several years now, since version 2.2, and plan to continue expanding their usage. Most importantly, they’ll soon merge the batch and streaming layers into a single, authoritative streaming layer.

Throughout this project, the Twitter team collaborated very closely with Google engineers to exchange feedback and discuss product enhancements. We look forward to continuing this joint technical effort on several ongoing large-scale cloud migration projects at Twitter. Stay tuned for more updates!