$validate checks each incoming document in an Atlas Stream Processing pipeline against a user-defined schema. Use it when downstream processing or storage requires particular fields, data types, values, or ranges. Documents that fail validation do not proceed to the next stage. section 15.5.2 section 15.6.4
{
"$validate": {
"validator": { <filter> }, #1
"validationAction": "discard" | "dlq" #2
}
}
The stage has two settings: validator holds the conditions, and validationAction determines what happens when a document fails them. section 15.5.2
To add validation to a processor:
- Start the pipeline with $source, then place $validate anywhere before the final $emit or $merge stage. section 15.5.2 section 15.5
- Put the required field checks in validator. It supports query operators except $near, $nearSphere, $text, and $where. section 15.5.2
- Choose validationAction: discard drops invalid documents and is the default; dlq logs them to a dead-letter queue while discarding them, without guarantees. section 15.5.2
Validate records before storing them
A processor can read from a MongoDB collection, validate that name exists and is a string, age exists and is an integer of at least 18, and email matches a pattern, then send passing documents to a destination collection. With validationAction set to discard, failing records never reach that destination. section 15.6.4
$validate is optional. Add it when rejecting malformed stream data is preferable to letting that data reach later transformations or a sink. If failed documents need further handling, consider dlq, but do not treat its logging as guaranteed delivery. section 15.5 section 15.5.2
