How Qubit deduplicates streaming data at scale with Google Cloud Platform

At Qubit, we enable over 400 million personalized experiences per month, and to power those experiences we process data from websites, mobile apps and other customer touchpoints. Managing and processing this data in near-real time is a significant challenge; various stages of the pipeline process events at over 500K messages per second, all of which need to be validated, enriched and routed with an overall latency of a few seconds end to end.

Since Qubit retrieves data from multiple client environments (mobile, server-to-server, web), we end up being exposed quite dramatically to the challenges of real-world data collection. One of these, in particular, is duplicated data.

Duplicates may occur for numerous reasons. One of the most common is client-side retries, which can occur when mobile or web clients re-send events because they cannot confirm that the previous request succeeded. Exactly-once message delivery — hard in the best of times — is even harder on the open internet. As a consequence, in a number of cases (and particularly over flaky internet conditions), data records may end up being received by our data processing pipeline multiple times.

Note that this problem differs from the exactly-once processing challenge in that even if you process each duplicate message exactly once, you still have a problem. Most real-time data pipelines are —​ very reductively —​ counting things, so when messages are duplicated, these counts go awry. For example, a sum of total amount of products purchased would be incorrect if computed naively across duplicated event data because of double (or even multiple!) counting. This presents a problem: duplicated events may throw off calculations and make inferences on the data inaccurate, or possibly even completely wrong.

Toward scoped deduplication

Our ideas rapidly crystallized into the following goals:

• Provide a single, performant duplicate tagger that detects messages sent multiple times to our APIs
• Make it easy for developers to add granular, application-specific deduplication to their own systems (for when duplicates may be generated for reasons other than multiple sends from clients)

Choosing a storage system

As we're happy users of Google Cloud Bigtable, it was a natural choice for storing a large dataset that requires low-latency access even under relatively large, spiky data throughput requirements and high-availability demands.

Designing a schema

This is an efficient representation for the data: checkpoint cells are created as and when records are actually processed by applications. The hashing of the id will ensure that keys are well distributed across tablet servers, which helps avoid hotspots regardless of the nature of the original id. Additionally, often engineers like to store small amounts of metadata —​ such as the time when the record was processed, for instance —​ and we allow this metadata to be stored alongside the checkpoint data.

Since the row tends to grow horizontally, i.e., we add a cell when a new application checkpoint arrives, with a new cell created per checkpoint, this is what we call a fat table. In such tables, there are only as many rows as there are ids, but a large number of cells (in this case, across a single column family). However, after many years of working with Apache HBase (which is an open source implementation of Bigtable), we learned that skinny tables, which are taller than they are wide, are usually much easier to manage in a high-performance way as new use cases emerge for tables.

So, a skinny transposition of the schema looks something like this:

We generate a fixed length rowkey — something we tend to try to do with most of our schemas — which is a concatenation of the hash of message id with a hash of the application identifier (typically its name). By using hashes of the ids and the app name, we get a fixed-length rowkey that allows efficient scanning of sub-ranges. In this case, we have at least as many rows as there are checkpoints, in contrast to the earlier schema where the cardinality of ids informed the upper bound on the number of rows.

An immediately obvious drawback is that this schema requires more storage, as each rowkey also stores a hash of the application name. This isn’t a total loss, though, for a couple reasons.

First of all, the identifier is a common and repeated key prefix across millions or billions of rows, which allows for efficient compression. More important, it's more than compensated by the fact that these rowkeys allow efficient range queries, particularly on a per-application basis. In practice, this identifier also often includes an internal client-tracking code, so that we generate a unique hash for each application and client combination.

These ranged scans are useful because they allow bounded scans (i.e., without a full table scan) to efficiently read all the previous ids an application has already processed. This approach is useful for large batch jobs — in disaster-recovery use cases, for example — that can download the full list of processed message ids instead of having to make individual point lookups.

Application users are also allowed to supply custom metadata (such as timestamps) that can be stored alongside the entry to record when, for example, the record was processed.

Optimizing checkpointing

A naive get followed by a put to insert the checkpoint is both relatively slow and error prone: there's an obvious race condition if the state is checked concurrently. Fortunately, Cloud Bigtable supports conditional mutations (changes are applied if a condition is met and rejected otherwise). We use this feature to checkpoint only if a checkpoint cell has not previously been created, rejecting the checkpoint otherwise — all done atomically. Users can now be informed if the message has been previously seen, and if not, they can be sure only a single worker is allowed to checkpoint.

Data processing with Apache Beam and Google Cloud Dataflow

While we bundle our dedupe system as a library (to make it easy to integrate into existing code in multiple environments), the primary dedupe via Cloud Dataflow is a trivial Beam DoFn.

(Protip: use AbstractCloudBigtableTableDoFn to help take care of the Cloud Bigtable connection lifecycle.)

Looking ahead, as state-management APIs in Beam are formalized, we intend to optimize this process further by detecting duplicated messages in our core flows via state held in the Cloud Dataflow service.

Next steps

In a future post, we’ll describe the analytic systems we built at Qubit to analyze this data at scale.