$emit is the final sink stage in an Atlas Stream Processing pipeline. It sends processed stream records to an external streaming destination, such as a Kafka topic, after source, validation, filtering, transformation, enrichment, or windowing stages have run. section 15.5 section 15.6.4
Use $emit when the processed records must remain in a streaming system rather than being written to MongoDB.
- Place it after the intermediate stages that process or transform incoming records. section 15.5
- Treat it as one of the pipeline's final destinations, alongside $merge for writing to a MongoDB collection. section 15.6.4
- Configure the sink connection and destination through the connection registry; placeholder connections do not run until properly defined. section 15.6.4
Forwarding processed Kafka records
A stream processor can ingest records from a source, apply stages such as $match or $addFields, and use $emit to forward the resulting records to a Kafka topic for further use. section 15.6.4
Listing 15.2 The stream processor key components pseudocode
// A processor is essentially a sequence of stages organized in an array,
// and it can be constructed incrementally using variables within the
// mongosh shell. Right now, this is an empty processor, meaning nothing is
// defined yet.
let processor = [] // #1
// The first stage must always be a source. The source retrieves data from
// a connection, which is defined in the connection registry.
// For now, the source is empty.
let source = {$source:{}} // #2
// Add the source to the processor. This creates the first stage in the
// processor pipeline.
processor = [source] // #3
// At this point, the processor is useful to inspect the source data.
// We can preview the data in the shell by running .process() command.
// This will display the results continuously to the shell.
sp.process(processor) // #4
// The sink is the final stage. It specifies where the data should go, such
// as sending it to a MongoDB collection using $merge or to a Kafka topic
// using $emit.
let sink = {$merge: {}} // #5
// This simple processor transfers data from a source to a sink without
// modifying it.
processor = [source, sink] // #6
// Between the source and the sink, you can add any number of stages to
// process the data. These stages work like standard MongoDB aggregation
// stages, such as $match, $addFields, etc.
processor = [source, stageA, stageB, ..., sink] // #7
// Finally, processors can be saved with a name for future use.
// You can start or stop them as needed.
// When started, they run continuously in the background.
sp.createStreamProcessor(
➥’myProcessor’, processor); // #8
sp.myProcessor.start(); // #9
Inspect the stream processor skeleton to see how a source, intermediate stages, and a final sink are assembled and run continuously. section 15.6.4
