~/blog$ cat on-building-data-pi…

On Building Data Pipelines

Overview

In practice, most nontrivial applications have requirements that cannot be satisfied solely by an OLTP database like Postgres or MySQL. Consider an application that combines the following technologies: full-text index to handle search queries, distributed cache to speed up data retrieval for common requests, and data warehouse for analytics.

The same data is repeated in several places and needs to be kept in sync. If a record is updated in the database, it will also need to be updated in the search index, cache, and data warehouse.

In this post, I will describe three ways to synchronize data, namely batch processing, dual writes, and stream processing. I will describe how change data capture, a form of stream processing, builds durable, ordered, near real-time data pipelines.

Batch Processing

Data warehouses typically sync data using batch ETL (extract, transform, load processes). One way to do this would be to snapshot the database (extract) at some regular cadence, sanitize and clean the data (transform), and bulk load the data into the data warehouse.

ETL pipeline from PostgreSQL through extract, transform, and load stages into S3, queried by Athena for BI tools, ML models, and data analysts

Hourly ETL pipeline from PostgreSQL to S3, queried by Athena for downstream consumers

Dual Writes

If batch processing is too slow for your application’s requirements, dual writes is an alternative pattern, in which application code explicitly writes to two (or more) systems. However, there are serious risks involved with dual writes because application code is writing to several systems outside of a single transaction.

Dual writes are prone to race conditions which can lead to data inconsistency. Consider an application that updates its search index immediately after updating the database. Two clients concurrently want to update record Foo in the database, then propagate that update to the search index. Due to unfortunate timing, the requests are interleaved. Client 1 updates record Foo with value X, then client 2 updates record Foo with value Y. Client 2 then updates the search index with value Y and later client 1 updates the search index with value X. The final value for record Foo in the database is X, but Y in the search index.

Sequence diagram showing two clients racing to update a database and search index, resulting in the database storing Y but the index storing X

Race condition in dual writes: interleaved requests leave the database and search index out of sync

Another problem is that one write may fail while the other succeeds, leading to inconsitent data. Without a distributed transaction, a client may successfully update a record in the database but fail to update the search index altogether.

Sequence diagram showing a client successfully updating the database but receiving an error when writing to the search index, leaving them inconsistent

Partial failure in dual writes: the database write succeeds but the search index write fails

Finally, dual writes complicate business logic, requiring code to be aware of downstream systems.

Stream Processing

Stream processing offers a middle ground: lower latency than batch processing without many of the consistency risks of dual writes. The idea is to stream data updates as changes are made to the leader database and make those data updates available to follower systems.

Change Data Capture

With the architecture of many distributed systems consisting of a single source-of-truth database for user-generated writes and derived data stores for handling more specific requirements, change data capture (CDC) has emerged as a popular stream processing solution for data replication. CDC is the process of observing and extracting all data changes written to a database and making those changes available to other systems for data replication.

CDC is implemented by parsing the database’s transaction logs for changes and publishing those changes to a system of record for downstream consumers. The system of record is typically a durable stream like Kafka or AWS Kinesis. A key property of durable streams is the ability to replay history, which is critical for consumers to prevent data loss in the event of unexpected failure.

Why not capture changes with database triggers?

Because triggers run in the write path, a transaction cannot commit until the trigger completes. If the trigger publishes to Kafka or Kinesis, every write now pays network latency and depends on an external system being available. Parsing transaction logs, by contrast, is asynchronous, out-of-band, and replayable.

CDC architecture: a CDC connector observes the PostgreSQL write-ahead log and publishes events to a message broker, which fans out to Elasticsearch, Redis, and a data warehouse

Change data capture pipeline: transaction log changes are published to a message broker and consumed by downstream data stores

Examples of CDC in the Wild

Building a CDC Pipeline with Kinesis and Lambda

Kinesis Concepts

Much of the helpful terminology is taken from the Kinesis Terms and Concepts docs. I’m copying a subset of it here for convenience.

  • Kinesis Data Stream: A Kinesis data stream is a set of shards. Each shard has a sequence of data records. Each data record has a sequence number that is assigned by Kinesis Data Streams.
  • Data record: A data record is the unit of data stored in a Kinesis data stream. Data records are composed of a sequence number, a partition key, and a data blob, which is an immutable sequence of bytes.
  • Shard: A shard is a uniquely identified sequence of data records in a stream. You can think of a shard as an ordered queue within a stream and a stream consists of many shards. It’s important to recognize that order only is guaranteed at the shard-level. Each shard can support up to 5 transactions per second for reads, up to a maximum total data read rate of 2 MB per second and up to 1,000 records per second for writes, up to a maximum total data write rate of 1 MB per second. If your data rate increases, you can increase or decrease the number of shards allocated to your stream. Kinesis auto-scaling works by splitting and merging shards.
  • Partition Key: A partition key is used to group data by shard within a stream. Kinesis Data Streams segregates the data records belonging to a stream into multiple shards. It uses the partition key that is associated with each data record to determine which shard a given data record belongs to. One shard can consist of several partition keys.

Choosing the Right Partition Key

Because shards have limited write and read throughput, it’s recommended to use high-cardinality partition key values like a primary key. This way, data records will be uniformly distributed across all the shards in a stream, preventing hot and cold shards.

Message Delivery Semantics in Distributed Systems

Message delivery semantics in distributed systems are behavioral guarantees that dictate how data is shared and perceived.

They're like contracts agreed upon by the producer, message broker, and consumer.

There are 3 types of message delivery semantics:

  1. At-most-once. Messages are delivered either once or not at all. We’re okay with losing data to avoid the possibility of duplicate data.
  2. At-least-once. Messages are delivered once or multiple times. We’re okay with duplicate data.
  3. Exactly-once. Messages are delivered exactly once. We're okay with solving theoretically impossible problems.

Exactly-once is exactly-the-reason why distributed systems engineers lose sleep. For CDC, at-least-once is the right choice.

Designing for At-Least-Once Delivery in Kinesis

Consider a producer that unexpectedly terminates after calling PutRecord, but before it receives acknowledgement from Kinesis. Assuming that data loss is unacceptable in this application, the producer retries and receives acknowledgement for the retry. Suppose both writes were successfully committed to Kinesis, there will be two identical records with unique sequence numbers.

For mutable downstream stores like Redis or DynamoDB, duplicate writes are usually safe because upserts are naturally idempotent.

For append-only stores like audit logs or financial ledgers, duplicate writes are unacceptable. In those cases, include an idempotency key in each record and deduplicate before writing to the downstream store.

Lambda as a Kinesis Consumer

When configuring custom consumers, you have to manage connections per shard, iterators to keep track of where the consumer is reading from the shard, and the consumer’s polling strategy. With Lambda consumers, the Lambda service will manage these details on your behalf. Here’s a diagram of how it works:

Diagram showing data flowing from a source into Kinesis shards, each mapped 1:1 to a Lambda poller and function, fanning out to two destinations

Lambda automatically scales one poller and function per Kinesis shard, fanning out to multiple destinations

In the default configuration, each Lambda consumer (only one consumer depicted above) spawns one Lambda function instance per shard. And each shard shares its 2 MB per second read throughput amongst all consumers. These limitations can be improved by enabling enhanced fan-out and parallelization factor.

Dedicated Throughput with Enhanced Fan-Out

With enhanced fan-out (EFO), each consumer gets its own dedicated 2MB per second connection. This allows you to add multiple consumers without throttling your shard. However, there are additional costs to running EFO consumers. That being said, enhanced fan out is not necessary in the aforementioned design where there is a 1:1 mapping between data stream and consumer.

Concurrent Processing with Parallelization factor

Parallelization factor can be set to increase concurrent Lambda invocations per shard. For example, with a parallelization factor of 2, there can be up to 2 concurrent Lambda instances processing a single shard. Parallelization at the shard level is achieved by grouping data records by partition key at “read-batch” time. Each grouping is polled by a lambda instance for processing. This results in order no longer being maintained at the shard level, but rather the partition key level.

Stream poller architecture: a record processor reads from a shard, consistent hashing routes records to batchers grouped by partition key, and each batcher invokes a separate Lambda function

Parallelization factor: consistent hashing partitions records from a single shard into batchers, each invoking a separate Lambda function

Why don't we need shard-level order guarantees?

In practice, preserving order at the row level is usually enough. Using the row's primary key as the partition key gives you that guarantee.

Consider a database that processes the following transactions in this order:

  1. Insert row A into Table X
  2. Insert row B into Table Y
  3. Update row A

As long as step 1 and 3 are consumed in order, it’s not important whether step 2 is consumed before or after steps 1 and 3. Row-level order guarantees can be achieved by using the primary key as the partition key. This feature is supported out of the box by AWS Database Migration Service (DMS).

Isolating Poison Pills with BisectBatchOnFunctionError

If a function invocation returns an error, Lambda will recursively split the batch in half and retry. This feature allows you to separate bad records from good ones but comes with the consequence that records may be processed more than once. Records are still guaranteed to be processed in order.

Records that remain problematic can be sent to a dead letter queue after they exceed the maximum record age or the maximum number of retry attempts.