Pipeline Patterns Reference

Official examples repo: https://github.com/mongodb/ASP_example (quickstarts, example processors, Terraform examples). Start with example_processors/README.md for the full pattern catalog. Always consult the official repo for the latest validated patterns before creating processors.

Stage Quick-Reference

Invalid Constructs

Do NOT use these in streaming pipelines:

• $$NOW, $$ROOT, $$CURRENT — not available in stream processing
• HTTPS connections as $source — HTTPS is for $https enrichment only
• Kafka $source without topic — topic field is required
• Pipelines without a sink — $merge/$emit required for deployed processors (sinkless only works via sp.process())
• Lambda connections with $emit — Lambda uses $externalFunction (can be mid-pipeline or terminal sink with async execution), not $emit

Source Patterns

MongoDB Change Stream

{"$source": {"connectionName": "my-cluster"}}

With full document and pushdown pipeline:

{"$source": {
  "connectionName": "my-cluster",
  "db": "mydb", "coll": "mycoll",
  "fullDocument": "updateLookup",
  "fullDocumentBeforeChange": "whenAvailable",
  "pipeline": [{"$match": {"operationType": "insert"}}]
}}

Kafka (topic is REQUIRED)

{"$source": {
  "connectionName": "my-kafka",
  "topic": "my-topic",
  "auto_offset_reset": "earliest",
  "partitionIdleTimeout": {"size": 30, "unit": "second"}
}}

Kinesis

{"$source": {
  "connectionName": "my-kinesis",
  "stream": "my-stream",
  "config": {"initialPosition": "TRIM_HORIZON"},
  "shardIdleTimeout": {"size": 30, "unit": "second"},
  "consumerARN": "arn:aws:kinesis:us-east-1:123456789:stream/my-stream/consumer/my-consumer:123"
}}

stream (required): Kinesis stream name. config.initialPosition: TRIM_HORIZON (oldest, default) or LATEST. shardIdleTimeout: unblocks windows when shards go idle (like Kafka partitionIdleTimeout). consumerARN (optional): enables enhanced fan-out for dedicated throughput.

Inline Documents (ephemeral testing only)

{"$source": {"documents": [{"device_id": "sensor-1", "temp": 72.5}]}}

Sink Patterns

$merge to Atlas

{"$merge": {"into": {"connectionName": "my-atlas", "db": "mydb", "coll": "mycoll"}}}

With match behavior and parallelism:

{"$merge": {
  "into": {"connectionName": "my-atlas", "db": "mydb", "coll": "mycoll"},
  "on": "_id", "whenMatched": "replace", "whenNotMatched": "insert",
  "parallelism": 4
}}

whenMatched: replace, merge, delete (via $cond). whenNotMatched: insert.

Additive merge (append to arrays):

{"$merge": {
  "into": {"connectionName": "my-atlas", "db": "mydb", "coll": "mycoll"},
  "on": "device_id",
  "whenMatched": [{"$addFields": {"readings": {"$concatArrays": ["$readings", "$$new.readings"]}}}],
  "whenNotMatched": "insert"
}}

Dynamic routing:

{"$merge": {"into": {
  "connectionName": "my-atlas", "db": "mydb",
  "coll": {"$cond": {"if": {"$eq": ["$priority", "high"]}, "then": "alerts", "else": "events"}}
}}}

$emit to Kafka

{"$emit": {
  "connectionName": "my-kafka", "topic": "output-topic",
  "key": {"field": "device_id", "format": "string"}
}}

Key formats: string, json, int, long, binData. Tombstone support: "tombstoneWhen": {"$expr": {"$eq": ["$status", "deleted"]}}.

$emit to Kafka with Schema Registry (Avro)

{"$emit": {
  "connectionName": "my-kafka", "topic": "output-topic",
  "schemaRegistry": {
    "connectionName": "my-schema-registry",
    "valueSchema": {
      "type": "avro",
      "schema": {
        "type": "record", "name": "SensorReading",
        "fields": [
          {"name": "device_id", "type": "string"},
          {"name": "temp", "type": "double"},
          {"name": "timestamp", "type": "long"}
        ]
      },
      "options": {
        "subjectNameStrategy": "TopicNameStrategy",
        "autoRegisterSchemas": true
      }
    }
  }
}}

Requires a SchemaRegistry connection (see connection-configs.md). valueSchema.type must be lowercase avro (case-sensitive). valueSchema.schema is always required, even with autoRegisterSchemas: true.

$emit to Kinesis

{"$emit": {"connectionName": "my-kinesis", "stream": "out", "partitionKey": "$device_id"}}

$emit to S3

{"$emit": {
  "connectionName": "my-s3", "bucket": "my-bucket",
  "path": {"$concat": ["data/", {"$dateToString": {"format": "%Y/%m/%d", "date": "$timestamp"}}]},
  "config": {"outputFormat": "relaxedJson"}
}}

Fields: connectionName (required), bucket (required), path (required — key prefix string or expression), region (optional), config (optional — outputFormat, writeOptions, delimiter, compression).

$https as Sink (webhook/API)

{"$https": {
  "connectionName": "my-webhook",
  "path": "/events",
  "method": "POST",
  "onError": "dlq"
}}

When used as a final sink stage, $https sends processed documents to an external HTTP endpoint. Unlike mid-pipeline usage (which enriches documents with API responses), sink usage doesn't expect a response to merge back into the document. Useful for:

• Sending data to webhooks
• Posting to external APIs
• Triggering external systems

$externalFunction as Sink (Lambda async)

{"$externalFunction": {
  "connectionName": "my-lambda",
  "functionName": "arn:aws:lambda:us-west-1:123456789:function:my-function",
  "execution": "async",
  "onError": "dlq"
}}

Important: When used as a final sink stage, $externalFunction MUST use execution: "async". This fires off the Lambda function without waiting for a response, useful for:

• Triggering downstream AWS applications or analytics
• Notifying external systems
• Firing off alerts or billing logic
• Propagating data to external workflows

Unlike mid-pipeline usage (where execution: "sync" is allowed for enrichment), sink usage requires async execution only. The pipeline still needs this as the terminal stage — you cannot use $emit to invoke Lambda.

Window Patterns

Tumbling

{"$tumblingWindow": {
  "interval": {"size": 5, "unit": "minute"},
  "pipeline": [{"$group": {"_id": "$deviceId", "avg": {"$avg": "$temp"}, "count": {"$sum": 1}}}]
}}

Hopping (with allowedLateness)

{"$hoppingWindow": {
  "interval": {"size": 5, "unit": "minute"},
  "hopSize": {"size": 1, "unit": "minute"},
  "allowedLateness": {"size": 15, "unit": "second"},
  "pipeline": [{"$group": {"_id": "$region", "total": {"$sum": "$amount"}}}]
}}

Session

{"$sessionWindow": {
  "gap": {"size": 5, "unit": "minute"}, "key": "$userId",
  "pipeline": [{"$group": {"_id": "$userId", "actions": {"$push": "$action"}, "count": {"$sum": 1}}}]
}}

Late data

{"$tumblingWindow": {
  "interval": {"size": 1, "unit": "minute"},
  "allowedLateness": {"size": 30, "unit": "second"},
  "boundaryType": "eventTime",
  "pipeline": [{"$group": {"_id": "$sensorId", "max": {"$max": "$value"}}}]
}}

boundaryType: eventTime (document timestamp) or processTime (wall clock, default).

Windowing Rules

• Windows require $group inside the window pipeline
• Idle Kafka partitions block windows — use partitionIdleTimeout
• allowedLateness lets late docs update closed windows

Enrichment Patterns

$https

{"$https": {
  "connectionName": "my-api",
  "path": {"$concat": ["/users/", "$userId"]},
  "method": "GET", "as": "userInfo", "onError": "dlq"
}}

onError: dlq (recommended), discard, fail. Store auth in connection settings, not pipeline. Place $https after windows to batch requests.

$lookup

{"$lookup": {
  "connectionName": "my-atlas",
  "from": {"db": "mydb", "coll": "users"},
  "localField": "userId", "foreignField": "_id", "as": "user",
  "parallelism": 2
}}

$externalFunction (Lambda - Mid-Pipeline Enrichment)

{"$externalFunction": {
  "connectionName": "my-lambda",
  "functionName": "my-function-name",
  "execution": "sync",
  "as": "lambdaResult",
  "onError": "dlq",
  "payload": [
    {"$project": {"userId": 1, "data": 1}}
  ]
}}

Mid-pipeline usage:

• execution: sync (waits for Lambda result, stores in as field) or async (non-blocking)
• as: Field name to store Lambda response (required for sync, ignored for async)
• payload: Optional inner pipeline to customize request body sent to Lambda
• Use for enriching/transforming documents before downstream stages

Sink usage: See the Sink Patterns section. When used as final stage, MUST use execution: "async" only.

$validate (Schema Validation)

{"$validate": {
  "validator": {"$jsonSchema": {
    "required": ["device_id", "timestamp", "reading"],
    "properties": {
      "device_id": {"bsonType": "string"},
      "reading": {"bsonType": "double"}
    }
  }},
  "validationAction": "dlq"
}}

validationAction: "dlq" (recommended), "discard", "error" (crashes processor — avoid in production). Place early to catch bad data before expensive stages.

$function (JavaScript UDF)

{"$addFields": {
  "boostedWatts": {"$function": {
    "body": "function(watts) { return watts * 1.2; }",
    "args": ["$watts"],
    "lang": "js"
  }}
}}

Requires SP30+ tier. body: JavaScript function as string. args: array of field references. lang: always "js".

Common Pipeline Patterns

Array Normalization

[
  {"$source": {"connectionName": "my-kafka", "topic": "orders"}},
  {"$unwind": "$items"},
  {"$replaceRoot": {"newRoot": {"$mergeObjects": ["$items", {"orderId": "$orderId", "ts": "$timestamp"}]}}},
  {"$merge": {"into": {"connectionName": "my-atlas", "db": "mydb", "coll": "line_items"}}}
]

Dynamic Kafka Topic Routing

{"$emit": {
  "connectionName": "my-kafka",
  "topic": {"$switch": {
    "branches": [
      {"case": {"$eq": ["$severity", "critical"]}, "then": "alerts-critical"},
      {"case": {"$eq": ["$severity", "warning"]}, "then": "alerts-warning"}
    ],
    "default": "alerts-info"
  }}
}}

Complex Event Processing (Fraud Detection)

[
  {"$source": {"connectionName": "my-kafka", "topic": "transactions"}},
  {"$tumblingWindow": {
    "interval": {"size": 5, "unit": "minute"},
    "pipeline": [
      {"$group": {
        "_id": "$userId",
        "txnCount": {"$sum": 1},
        "totalAmount": {"$sum": "$amount"},
        "uniqueLocations": {"$addToSet": "$location"}
      }},
      {"$addFields": {
        "suspiciousLocations": {"$gt": [{"$size": "$uniqueLocations"}, 3]},
        "highVelocity": {"$gt": ["$txnCount", 10]}
      }},
      {"$match": {"$or": [{"suspiciousLocations": true}, {"highVelocity": true}]}}
    ]
  }},
  {"$merge": {"into": {"connectionName": "my-atlas", "db": "fraud", "coll": "alerts"}}}
]

Graceful Degradation with $ifNull

{"$addFields": {
  "userName": {"$ifNull": ["$userInfo.name", "unknown"]},
  "userTier": {"$ifNull": ["$userInfo.tier", "standard"]},
  "enrichmentSucceeded": {"$ne": [{"$type": "$userInfo"}, "missing"]}
}}

Window Metadata

Inside window pipelines, _stream_meta.window.start and _stream_meta.window.end provide boundary timestamps:

{"$group": {
  "_id": "$deviceId",
  "windowStart": {"$first": "$_stream_meta.window.start"},
  "windowEnd": {"$first": "$_stream_meta.window.end"},
  "avg": {"$avg": "$temp"}
}}

Checkpoint Resume Constraints

With resumeFromCheckpoint: true (default), you CANNOT change: window type, interval, remove windows, or modify $source. Set false to make these changes (restarts from beginning).

DLQ Configuration

{"dlq": {"connectionName": "my-atlas", "db": "streams_dlq", "coll": "failed_documents"}}

DLQ documents include: original document, error message, stage info, timestamp.

Sample Stream Formats

Chained Processors (Multi-Sink Pattern)

CRITICAL: A single pipeline can only have ONE terminal sink ($merge or $emit). You CANNOT have both $merge and $emit as terminal stages. When a user requests multiple output destinations (e.g., "write to Atlas AND emit to Kafka" or "archive to S3 AND send to Lambda"), you MUST:

1. Acknowledge the single-sink