MongoDB Atlas Stream Processing

This skill manages MongoDB Atlas Stream Processing (ASP) pipelines using the sp CLI tool. Use natural language to create, deploy, monitor, and optimize streaming data workloads.

Running Commands

Use this path pattern for all sp commands:

~/.claude/plugins/cache/asp-skill/asp/*/tools/sp/sp <command>

Quick Reference

Setup Requirements

Before using, ensure the plugin has a valid config.txt in its directory:

PUBLIC_KEY=your_atlas_public_key
PRIVATE_KEY=your_atlas_private_key
PROJECT_ID=your_atlas_project_id
SP_WORKSPACE_NAME=your_workspace_name

Copy from config.txt.example and fill in your Atlas API credentials.

Core Commands

Workspace Management

Processor Management

Connection Management

Performance Profiling

Collection Management (Data Verification)

Materialized Views

Creating Processors

IMPORTANT: Before creating any processor, consult the official MongoDB ASP examples repo for best practices and valid patterns:

https://github.com/mongodb/ASP_example

This repo contains the latest source/sink configurations, valid pipeline stages, and working examples. Always fetch and review examples from this repo when:

• •Creating new processors
• •Configuring sources (Kafka, Atlas change streams, etc.)
• •Setting up sinks ($merge targets)
• •Using advanced features ($lookup, $function, windowing)

Processors are JSON files defining streaming pipelines:

{
    "name": "my_processor",
    "pipeline": [
        {
            "$source": {
                "connectionName": "my_connection",
                "timeField": { "$dateFromString": { "dateString": "$timestamp" }}
            }
        },
        {
            "$match": { "value": { "$gt": 100 } }
        },
        {
            "$merge": {
                "into": {
                    "connectionName": "Cluster01",
                    "db": "mydb",
                    "coll": "output"
                }
            }
        }
    ]
}

Pipeline Stages

• •$source - Data source (HTTP, Kafka, cluster)
• •$match - Filter documents
• •$project - Select/transform fields
• •$addFields - Add computed fields
• •$group - Aggregate data
• •$window - Window-based computations
• •$function - Custom JavaScript transformations
• •$merge - Write to destination

$source Configuration

Always consult https://github.com/mongodb/ASP_example for current source configurations.

Key points:

• •Kafka sources REQUIRE a topic field
• •HTTP connections are NOT sources - they are for lookups/enrichment only
• •Not all connection types are valid sources

Fetch examples from the ASP_example repo before creating processors to ensure correct syntax.

IMPORTANT: Invalid Variables

$$NOW, $$ROOT, $$CURRENT and similar MongoDB aggregation system variables are NOT valid in Atlas Stream Processing pipelines. Do not use them.

For timestamps, use the $source timeField or pass timestamps from your data source.

Tier Selection

Use --auto to let the tool analyze and select the optimal tier.

Common Workflows

Deploy a New Processor

1. •Create processor JSON in processors/
2. •sp processors create -p <name>
3. •sp processors start -p <name> --auto
4. •sp processors stats -p <name> - Check for DLQ errors
5. •sp collections count -c <db>.<output_collection> - Verify data is flowing

Verify Data Flow

After starting a processor with $merge, verify data reaches the destination:

1. •sp processors stats -p <name> - Check inputMessageCount > 0 and dlqMessageCount = 0
2. •sp collections count -c mydb.output - Confirm documents in target collection
3. •sp collections query -c mydb.output -l 5 - Sample documents to verify structure

If count is 0 but stats show input:

• •Check DLQ for errors: sp collections query -c <dlq_db>.<dlq_coll> -l 10
• •Verify $merge connectionName matches a valid cluster connection
• •Check field mappings in pipeline stages

Troubleshoot Performance

1. •sp processors list - Check status
2. •sp processors stats -p <name> --verbose - Get detailed stats
3. •sp processors profile -p <name> --duration 300 - Profile
4. •sp processors tier-advise -p <name> - Get tier advice

Update a Processor

1. •sp processors stop -p <name>
2. •Edit the JSON file
3. •sp processors drop -p <name>
4. •sp processors create -p <name>
5. •sp processors start -p <name> --auto

Reference Documentation

The docs/ directory contains comprehensive guides:

• •PIPELINE_PATTERNS.md - Common pipeline patterns and examples
• •CONNECTION_GUIDE.md - Connection types and configuration
• •PROCESSOR_SIZING_GUIDE.md - Tier selection and sizing
• •SP_USER_MANUAL.md - Complete sp CLI reference
• •ATLAS_STREAM_PROCESSING_PARALLELISM.md - Parallelism configuration
• •TESTING_GUIDE.md - Testing and validation

Consult these docs when building complex processors.

Output Format

All commands output JSON for easy parsing:

{
  "timestamp": "2024-01-15T10:30:00Z",
  "operation": "start_processor",
  "status": "success",
  "processor": "my_processor",
  "tier": "SP10"
}