DEV Community

MongoDB profile image MongoDB Guests
MongoDB Guests for MongoDB

Posted on

Streaming MongoDB Change Events to Amazon Redshift via Atlas Stream Processing and Kinesis

#mongodb #aws #ai

This tutorial was written by Igor Alekseev.

Real-time data pipelines that move operational data into analytical systems are a cornerstone of modern architectures. In this post, I’ll walk through how to stream MongoDB change events into Amazon Redshift using MongoDB Atlas Stream Processing and AWS Kinesis Data Streams — with no S3 staging or Firehose in between.

Architecture Overview

The flow is straightforward:

  1. MongoDB Atlas — Changes (inserts, updates, replaces) occur in a collection.
  2. Atlas Stream Processing — A stream processor captures these changes via a change stream, filters and reshapes them into flat JSON, and emits records to Kinesis.
  3. AWS Kinesis Data Stream — Acts as the durable, ordered buffer between the source and the analytical store.
  4. Amazon Redshift — A streaming materialized view reads directly from Kinesis and exposes the data as queryable columns.

The key advantage here is simplicity: there’s no intermediate S3 bucket, no Firehose delivery stream, and no ETL job to manage. Data flows from operational writes to analytical queries in near real-time.

Why This Combination?

Atlas Stream Processing handles the hardest part of change data capture — it connects natively to MongoDB change streams, handles resume tokens, and provides a pipeline language (aggregation framework) for reshaping documents on the fly. Instead of writing and hosting a custom consumer, you declare a pipeline, and Atlas manages the execution.

Kinesis provides a durable, ordered stream that decouples the producer (Atlas) from the consumer (Redshift). If Redshift has a maintenance window or you need to replay data, Kinesis retains records for up to 365 days.

Redshift Streaming Ingestion eliminates the traditional pattern of landing data in S3 first. A materialized view reads directly from the Kinesis stream, making new records queryable within seconds of arrival.

Implementation

Step 1: Create the Kinesis Data Stream 

aws kinesis create-stream \
  --stream-name atlas-change-events \
  --shard-count 1 \
  --region us-east-1
Enter fullscreen mode Exit fullscreen mode

A single shard handles up to 1 MB/s or 1,000 records/s ingest. Scale shards based on your change event volume.

Step 2: Set Up IAM for Atlas

Atlas Stream Processing authenticates to Kinesis via IAM AssumeRole. Create a role with a trust policy that allows the Atlas AWS account to assume it, and attach minimal Kinesis write permissions:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "kinesis:PutRecords",
        "kinesis:DescribeStreamSummary"
      ],
      "Resource": "arn:aws:kinesis:us-east-1:<YOUR_ACCOUNT>:stream/atlas-change-events"
    }
  ]
}
Enter fullscreen mode Exit fullscreen mode

That’s it — just two permissions. Atlas uses the batch API (PutRecords) exclusively, so you don’t need the single-record API (PutRecord) or the full DescribeStream.

Step 3: Register the Kinesis Connection in Atlas

In the Atlas UI, navigate to Stream Processing → Connection Registry → Add Connection → Kinesis, and provide:

  • The stream name
  • The AWS region
  • The IAM role ARN

Step 4: Create the Stream Processor

The stream processor is an aggregation pipeline that runs continuously. Here’s the pattern:

sp.createStreamProcessor("atlasToKinesisProcessor", [
  // 1. Filter for relevant operation types
  { $match: { operationType: { $in: ["insert", "update", "replace"] } } },

  // 2. Flatten the change event — promote fullDocument fields to top level
  { $replaceRoot: {
      newRoot: { $mergeObjects: ["$fullDocument", { _operationType: "$operationType" }] }
  }},

  // 3. Project into a flat structure for Kinesis/Redshift
  { $project: {
      device_id: "$device_id",
      timestamp: "$timestamp",
      watts: "$obs.watts",
      temp: "$obs.temp",
      _operationType: 1
  }},

  // 4. Emit to Kinesis
  { $emit: {
      connectionName: "myKinesisConnection",
      stream: "atlas-change-events",
      region: "us-east-1",
      partitionKey: "$device_id"
  }}
]);
Enter fullscreen mode Exit fullscreen mode

The $emit stage is Kinesis-specific. Note that stream, region, and partitionKey are all top-level fields — not nested inside a config object.

After creating the processor, start it in mongosh connected to your Stream Processing Instance (SPI):

sp.atlasToKinesisProcessor.start();
Enter fullscreen mode Exit fullscreen mode

Step 5: Set Up Redshift Streaming Ingestion

On the Redshift side, create an external schema pointing to Kinesis:

CREATE EXTERNAL SCHEMA IF NOT EXISTS kinesis_schema
FROM KINESIS
IAM_ROLE 'arn:aws:iam::<YOUR_ACCOUNT>:role/<REDSHIFT_ROLE>';
Enter fullscreen mode Exit fullscreen mode

Then create a streaming materialized view that parses the JSON records:

CREATE MATERIALIZED VIEW mdb_change_events AUTO REFRESH YES AS
SELECT
  approximate_arrival_timestamp,
  JSON_EXTRACT_PATH_TEXT(from_varbyte(kinesis_data, 'utf-8'), 'device_id') AS device_id,
  JSON_EXTRACT_PATH_TEXT(from_varbyte(kinesis_data, 'utf-8'), 'timestamp') AS event_timestamp,
  CAST(JSON_EXTRACT_PATH_TEXT(from_varbyte(kinesis_data, 'utf-8'), 'watts') AS DOUBLE PRECISION) AS watts,
  CAST(JSON_EXTRACT_PATH_TEXT(from_varbyte(kinesis_data, 'utf-8'), 'temp') AS DOUBLE PRECISION) AS temp,
  JSON_EXTRACT_PATH_TEXT(from_varbyte(kinesis_data, 'utf-8'), '_operationType') AS operation_type
FROM kinesis_schema."atlas-change-events";
Enter fullscreen mode Exit fullscreen mode

With AUTO REFRESH YES, Redshift continuously ingests new records. You can also trigger a manual REFRESH MATERIALIZED VIEW mdb_change_events; if needed.

Step 6: Query Your Data 

SELECT * FROM mdb_change_events
ORDER BY approximate_arrival_timestamp DESC
LIMIT 10;
Enter fullscreen mode Exit fullscreen mode

That’s it. Operational writes in MongoDB are now queryable in Redshift within seconds.

Lessons Learned

  • Flatten before emitting. Redshift’s JSON_EXTRACT_PATH_TEXT is limited to 5 levels of nesting, has a 16MB record cap, and AWS recommends migrating to JSON_PARSE with SUPER type instead. By flattening in Atlas Stream Processing ($replaceRoot + $project), Kinesis receives simple flat JSON, and the Redshift materialized view stays clean.
  • The materialized view only captures new records. Records already in the stream when the view is created may not appear. Create the view before starting the processor, or insert fresh test data after creation.
  • External schema creation requires a superuser. The CREATE EXTERNAL SCHEMA statement must be run by the database owner or a user with the CREATE SCHEMA privilege. If your IAM-federated session doesn’t have these privileges, connect with the admin user and password instead.

When to Use This Pattern

This architecture works well when:

  • You need near-real-time analytics on operational MongoDB data
  • You want to avoid managing CDC infrastructure (Debezium, Kafka Connect, custom consumers)i
  • Your analytical queries run in Redshift, and you want columnar performance
  • You prefer a serverless, managed approach with minimal operational overhead

For higher throughput or more complex transformations, you might add more Kinesis shards and scale the Atlas Stream Processing instance accordingly. The architecture stays the same — just the capacity changes.

References

Top comments (0)