Capturing Every Change From Shopify’s Sharded Monolith - Shopify

Shopify’s Move to Log-Based Change-Data Capture

Shopify’s data platform has evolved considerably since 2004, growing from a warehouse built for internal analytics into a data lake that supports over one million merchants. Today, the architecture is split into two main platforms: an internal data warehouse and a dedicated Merchant Analytics Platform.

• Internal Data Warehouse: Uses scheduled, batch-style query-based extraction. Change events from Shopify’s shards are written to Cloud Storage and denormalized into large Hive/Parquet tables. Spark and dbt handle transformations, with output loaded into sinks like BigQuery, Redshift, and custom serving applications.
• Merchant Analytics Platform: Runs an always-on batch-query application that extracts change events to Kafka. Lightweight streaming transformations write back to Kafka, which is then streamed into Google Bigtable for low-latency querying from merchant admin pages.

Running two separate platforms created practical problems. The toolchains are completely different — building for merchant analytics requires Kafka streaming applications in Golang, while internal analytics uses Spark and SQL-based batch jobs. Keeping two teams fluent in both stacks is costly. More subtly, the two extraction pipelines can return different results for the same table, since query timing and execution conditions affect outcome. That means duplication for every dashboard, report, or data science workflow that needs both platforms’ data.

As Shopify expanded into adjacent commerce offerings like Email, Capital, Balance, and Marketing, merchant analytics demand grew well beyond storefront reporting. With more sources feeding both platforms, the divergence problem gets worse. Unifying extraction into one pipeline is the only way to keep these systems coherent.

Three requirements drive that unified pipeline:

1. Access to application data: Most relevant data lives inside Shopify’s sharded monolith, which hosts the core business applications.
2. Freshness: Data must be available for general-purpose consumption as quickly as possible.
3. Accuracy and completeness: The capture mechanism must reflect every change in the upstream database without loss.

The rest of this post details how Shopify is rebuilding its extraction layer to meet those targets: moving from query-based CDC tooling (Longboat) to an immutable, append-only, log-based mechanism built on Kafka Connect and Debezium.

Why Batch Extraction Hits a Ceiling

Shopify’s core monolith is sharded across many MySQL instances, and getting data out of those stores has traditionally meant reaching in, grabbing what changed, and writing it somewhere else. That job belongs to Longboat, Shopify’s internal query-based change-data capture (CDC) service. Longboat runs batch jobs that periodically query source tables for rows whose "updated_at" column has changed since the last run, writing results to cloud storage for downstream analytics and reporting.

Longboat queries are written per table — no joins — and run against read-only replicas to reduce load. Even so, a given table can only be queried once an hour at most, and after adding query time, processing, and copying, the practical freshness floor is about one hour. That data then lands in storage and is assembled into snapshots for the data team’s batch workloads, but it’s rarely suitable as a general-purpose source for the rest of the company.

The Gaps in Query-Based CDC

Catching rows by a timestamp field is simple in principle: run SELECT * FROM `TABLE`, then narrow it down with "updated_at" cutoffs for incremental jobs. It works on tables with billions of rows, unlike a full table scan, but it has structural blind spots that surface as business requirements move toward near-real-time data.

Hard Deletes Vanish

Soft deletes — a flag marking a row as removed — update "updated_at" and are therefore captured. Hard deletes remove the row entirely, leaving no timestamp for a query to find. Query-based extraction misses every hard delete, so the data lake never learns about them. That absence has pushed many tables toward soft deletion, which is problematic when tables accumulate large volumes of obsolete rows just to keep CDC working; that hurts performance and raises data-retention concerns.

blinded by "updated_at" Bugs and Migrations

Query-based capture assumes the "updated_at" column is touched on every write. Two common cases violate that assumption. Large-scale data migrations deliberately avoid updating the timestamp so incremental queries aren't flooded with results. And simple coding errors can leave a row’s contents changed without bumping the timestamp. In both cases, the mutation is invisible to Longboat — an update from Peyton to Eli is simply lost downstream if nobody updates "updated_at".

Intermediate State Is Lost

Longboat reports only the most recent state of a row. If a row is updated several times between two job runs, every intermediate version is irretrievable. That’s a hard blocker for data modeling that needs the full sequence of a merchant’s or buyer’s actions.

Snapshots Don’t Agree on Time

Joining snapshots from different tables introduces a consistency problem. If the latest sales snapshot has a high-water mark of 2:15 and the products snapshot’s mark is 2:30, there’s no single answer to how fresh the joined result is. Communicating accurate freshness to stakeholders becomes guesswork.

Rows That Never Sit Still

In rare cases, a row is updated so frequently that every Longboat query finds it changed again by the time the extraction runs. The extractor effectively chases a moving target, and the row’s latest state may never be captured cleanly.

Speed Is Not the Answer

Raising Longboat’s polling frequency doesn’t fix these problems — it worsens the underlying cost. Each additional query adds pressure to the source datastore, and at some point running comprehensive scans fast enough becomes technically infeasible. Longboat is already at that edge.

Even if queries were faster, the rest of the pipeline remains batch-oriented: each step waits for the previous one to finish. A slow query, a resource-strapped processing cluster, or a network hiccup cascades into further delay. Batch systems can be tuned, but business needs are moving past what batch can deliver; streaming is the only path left.

Committing to Log-Based CDC

New requirements — real-time reporting, accurate business decisions, merchant-facing freshness — demand a source of comprehensive and timely data that Longboat can’t provide. Shopify already runs Apache Kafka as its distributed event broker, and purpose-built CDC tools such as Debezium and Maxwell’s Daemon exist to convert database changes into events. Moving to event streams also unblocks native event-driven applications elsewhere in the company, where the same fresh-data need has emerged independently.

Shopify’s MySQL topology runs with a primary instance and read replicas, and those replicas are synchronized by MySQL’s binary log — an internal record of every change to table data. That same log is the foundation for log-based CDC, so the path was clear to replace Longboat with a new platform built on proven event-streaming technology rather than reinventing the mechanism.

Choice of CDC Engine

A cross-functional team began evaluating log-based change-data capture in mid-2019, looking at Debezium, Maxwell’s Daemon, Spinal Tap, and a DIY approach along the lines of Netflix’s DBLog. Debezium came out ahead: it is the most active open-source CDC project, has responsive core maintainers, and supports a range of databases beyond MySQL, which matters given possible future non-MySQL sources.

Where Longboat extracts MySQL data to cloud storage, Debezium writes change records to Apache Kafka. Kafka is already central to Shopify’s infrastructure, so this fits existing tooling and gives consumers a standard stream-processing API.

Mapping Many Shards to One Topic Per Table

Out of the box, Debezium creates one topic per table per database instance. With more than 100 sharded databases, that would force consumers to read a separate topic for every shard. Shopify routes each shard’s events through a Kafka Connect RegexRouter transform into a single topic per shard (for example, shopify_shard_1_events), then a custom Kafka Streams application fans those records out to one topic per logical table.

All topics are partitioned by the source table’s primary key, so events for a given key land in the same partition. This topology isolates consumers from physical sharding.

The per-table output topics use Kafka log compaction: since CDC records are keyed by primary key, a compacted topic retains only the latest version of each key. Downstream consumers can therefore initialize a full local copy of an upstream table by reading the compacted topic, and storage needs stay proportional to the dataset’s key domain.

The platform meets near-real-time requirements: p99 latency from MySQL insert to Kafka availability is under 10 seconds. Median latency is far lower, though it can be affected by replication lag on the MySQL read replica that the Debezium connector reads from.

A single Debezium record looks like the example below, which shows one insertion into the addresses table:

Operational Lessons

Debezium only reached 1.0 in early 2020, and Shopify had no prior Kafka Connect usage. That combination meant encountering bugs first and building reliability practices as the platform grew.

Schemas and Evolution

Debezium supports Apache Avro schemas out of the box, with each captured table having a schema derived from its definition. Schema evolution rules allow additive changes, letting producers and consumers evolve independently. But some changes break compatibility—for instance, changing a column from Integer to String forces all consumers to update immediately or risk erroneous processing.

The tight coupling between external consumers and the source’s internal data model is an inherent drawback of CDC. Breaking changes and data migrations that are routine in application development ripple across many consumer codebases. Shopify works to avoid such changes, and is exploring ways to give consumers more warning when one is unavoidable.

Schema Registry

Confluent’s Kafka Schema Registry integrates cleanly with Debezium and existing Kafka consumers. It also serves as the basis for CDC data discovery: with one domain definition per topic, searching registered schemas reveals which topics hold which data. Combined with Kafka client identities and access control lists, this makes it possible to see which consumers read which data, generate dependency graphs, and track lineage of sensitive information.

Initial Table Snapshots

New Debezium connectors take an initial snapshot by running SELECT * FROM TABLE and writing the full table contents to Kafka before log-based CDC begins. This approach has significant limitations:

1. Snapshotting holds a read lock for the duration of each table snapshot. Snapshots can take hours, so they run against MySQL read replicas—still causing lock contention.
2. The table “allow.list” can be edited after the initial snapshot, but tables added later only receive change events from that point forward. Downstream consumers expect complete table state from the original snapshot, but later additions have none.
3. Debezium cannot snapshot a table without blocking incoming binlog events. For large tables that take hours to snapshot, that delay is unacceptable for latency-sensitive consumers.
4. Some Shopify core tables are simply too large to snapshot in any reasonable time.

These problems remain unresolved in the current platform, and early adopters have been told about them. An internal incremental snapshot tool has been proposed to address them.

Handling Large Records

Some Shopify workflows store very large text blobs in MySQL, reaching tens of MB—far beyond Kafka’s default 1 MB record limit. Raising the broker limit would hurt performance and still fail for future records that are larger again.

Two options were evaluated:

1. Splitting records into multiple Kafka events, recombined at consumption time.
2. External storage, where the Kafka record is a pointer to data held in Google Cloud Storage.

Splitting has the advantage of relying only on Kafka for storage and access control, and it was initially appealing. But it requires specialized consumers in every language used: typical consumers pass records straight to business logic, so substantial rework would be needed to guarantee correct recombination across batch boundaries. That seemed complex and error-prone.

External storage keeps production and consumption simple. Standard Kafka consumers can still read the records, though the payload is intentionally not directly accessible without the additional storage lookup. The downsides are reliance on a third-party store for availability and access control, plus the need to clean up GCS objects once their Kafka records are no longer available.

The chosen implementation is a custom Kafka serializer/deserializer (Serde) that encodes records as Avro via the Confluent Avro Converter. The Serde checks every record’s size: anything over the 1 MB limit is compressed, and if compression is still insufficient, the record is written to GCS and a placeholder is sent to Kafka, with the GCS path in the record header. Consumers deserialize through the same Serde, which handles decompression and GCS retrieval.

Large records reflect the historical decision to use MySQL as a blob store. The platform must support access to that data, but going forward Shopify wants to reduce dependence on shipping large payloads through Kafka—events should focus on a specific domain. GCS data must also be encrypted, gated by the same Kafka ACLs, and deleted when the corresponding Kafka record is removed, all of which adds complexity but is required for compliance.

Scale So Far

The CDC Kafka cluster stores over 400 TB of data. Roughly 150 Debezium connectors run across 12 Kubernetes pods. During the 2020 Black Friday Cyber Monday weekend, the system processed about 65,000 records per second on average, with spikes up to 100,000 per second—while only covering a subset of Shopify’s tables.

Adoption

One of the fastest-growing consumers is the Marketing Engagements API, which lets merchants see how buyers respond to campaigns. Moving from a batch warehouse model to Beam processing over a CDC stream improved data freshness from one day to one hour. Other teams are migrating batch applications off Longboat-generated tables and exploring stream-processing frameworks, including streaming joins and indefinitely materialized state.

Roadmap

Near-term focus areas include:

• Integrating CDC into streaming: With CDC data in Kafka, operational event streams and database changes share one source of truth. The team is evaluating Apache Beam, Kafka Streams, Apache Flink, and Materialize for rewriting important models as streaming pipelines.
• Integrating CDC into batch: The data platform is moving off its query-based engine to CDC event streams, resolving events into a batch format that meets existing expectations. The goal is a single source that all streaming and batch jobs consume from.
• Incremental snapshotting: The proposed tool would backfill chunks of existing MySQL tables without locking, reconciling incremental query results with ongoing binlog events to build current table state. This matters because several core tables are too large for the current lock-heavy snapshot approach.

Kafka-based CDC has become central to Shopify’s data platform modernization: standardizing extraction, replacing slow batch acquisition, and unifying batch and streaming sources. Every dataset change is now captured, reducing the delay between business events and actionable data.