A data source supplies the documents that enter a stream processor. Every Atlas Stream Processing pipeline begins with exactly one $source stage. The processor then handles incoming documents continuously, rather than running once like a regular query. section 15.4 section 15.9
For an external source, $source names a connection in the Connection Registry and identifies what to consume. Supported choices include a Kafka topic, a MongoDB collection change stream, and a MongoDB database change stream. Incoming documents must be valid JSON or Extended JSON (EJSON); EJSON can represent MongoDB values such as dates and ObjectIds. section 15.5.1 section 15.9
Choose and configure the source before adding processing stages:
- For Kafka, supply a registered connection name and topic. Optional settings in the $source prototype include timestamp mapping, partition idle timeout, and consumer configuration. section 15.5.1
- For a MongoDB change stream, supply the registered connection and database; identify a collection when consuming its change stream. The prototypes also show options for where to start and which change-document content to receive. section 15.5.1
- For a controlled test, supply a documents array instead of relying on unpredictable live input. This makes pipeline validation and troubleshooting easier. section 15.6.4 section 15.9
Start with a registered stream
In mongosh, a source can begin as `let s = {$source: {connectionName: "sample_stream_solar", timeField: {$dateFromString: {dateString: "$timestamp"}}}}`. The connection selects the incoming stream, while `timeField` maps its timestamp for stream processing. A complete processor can place `s` before a `$tumblingWindow` and `$merge`, so bounded aggregates are written to an Atlas collection. The connection must exist before this example can run. section 15.6.6 section 15.6.4
Listing 15.7 Using a document array as a source
// Create a source using an array of documents. ➥This array simulates incoming sensor messages let source = { $source: { "documents": [ { sensorName: 'sensor01', temperature: 22, humidity: 55, ➥timestamp: new Date("2023-09-07T12:00:00Z") }, { sensorName: 'sensor02', temperature: 45, humidity: 60, ➥timestamp: new Date("2023-09-07T12:01:00Z") }, { sensorName: 'sensor03', temperature: 5, humidity: 80, ➥timestamp: new Date("2023-09-07T12:02:00Z") } ] } } // #1 // Check for types and ranges of fields let validate = { $match: { sensorName: { $exists: true, $type: "string" }, temperature: { $exists: true, $type: "int", $gte: -50, // Temperature range from -50 to 100°C $lte: 100 }, humidity: { $exists: true, $type: "int", $gte: 0, // Humidity range from 0 to 100% $lte: 100 }, timestamp: { $exists: true, $type: "date" } } } // #2 // Projection with transformation of warning level using proper $cond let projectWithWarning = { $project: { sensorName: 1, temperature: 1, humidity: 1, timestamp: 1, warningLevel: { $cond: { if: { $gte: ["$temperature", 40] }, then: "HIGH", else: { $cond: { if: { $lte: ["$temperature", 10] }, then: "LOW", else: "NORMAL" } } } } } } // #3 // Filter only sensors with warnings let filterWarnings = { $match: { warningLevel: { $in: ["HIGH", "LOW"] } } } // #4 // Combine processing stages let processor = [source, validate, ➥ projectWithWarning, filterWarnings] // #5 // Execute the processor sp.process(processor) // #6
Inspect the document-array example for a fixed sensor-data source and the filtering stages that follow it. section 15.6.4
After $source, stages can validate, filter, transform, or enrich documents. Aggregations over multiple records need a time window to bound the otherwise continuous stream; a final stage sends results to a destination. To inspect source data while assembling a processor in the shell, place the source first and call `sp.process(processor)`. Placeholder connection names will not work until their connections are configured. section 15.5 section 15.6.4
