$emit Stage (Stream Processing)

string

Required

Name, as it appears in the Connection Registry, of the connection to ingest data from.

string | expression

Required

Name of the Apache Kafka topic to emit messages to.

document

Optional

Document containing fields that override various default values.

int

Optional

Number of acknowledgements required from the Apache Kafka cluster for a successful $emit operation.

The default value is all. Atlas Stream Processing supports the following values:

• -1

• 0

• 1

• all

string

Optional

Compression type for all data generated by the producer. The default is none (i.e no compression). Valid values are:

• none

• gzip

• snappy

• lz4

• zstd

Compression is used for full batches of data, so the efficacy of batching impacts the compression ratio; more batching results in better compression.

string

Optional

Date format for the date value. Valid values are:

• default - to use the default of the outputFormat.

• ISO8601 - to convert dates to strings in the ISO8601 format, which includes millisecond precision (YYYY-MM-DDTHH:mm:ss.sssZ).

For example:

Consider the following input.

{ "flightTime" : ISODate('2025-01-10T20:17:38.387Z') }

If $emit.config.defaultFormat is set to default, output looks similar to the following:

{ "flightTime" : {$date :"2025-01-10T20:17:38.387Z"}}

If $emit.config.defaultFormat is set to ISO8601, output looks similar to the following:

{ "flightTime" : "2025-01-10T20:17:38.387Z" }

expression

Optional

Headers to add to the output message. The expression must evaluate to either an object or an array.

If the expression evaluates to an object, Atlas Stream Processing constructs a header from each key-value pair in that object, where the key is the header name, and the value is the header value.

If the expression evaluates to an array, it must take the form of an array of key-value pair objects. For example:

[
 {k: "name1", v: ...},
 {k: "name2", v: ...},
 {k: "name3", v: ...}
]

Atlas Stream Processing constructs a header from each object in the array, where the key is the header name, and the value is the header value. Atlas Stream Processing supports header values of the following types:

• binData

• string

• object

• int

• long

• double

• null

object | string

Optional

Expression that evaluates to a Apache Kafka message key.

If you specify config.key, you must specify config.keyFormat.

string

Conditional

Data type used to deserialize Apache Kafka key data. Must be one of the following values:

• "binData"

• "string"

• "json"

• "int"

• "long"

Defaults to binData. If you specify config.key, you must specify config.keyFormat. If the config.key of a document does not deserialize successfully to the specified data type, Atlas Stream Processing sends it to your dead letter queue.

string

Optional

JSON format to use when emitting messages to Apache Kafka. Must be one of the following values:

• "relaxedJson"

• "canonicalJson"

Defaults to "relaxedJson".

string

Required

Name, as it appears in the Connection Registry, of the connection to write data to.

string

Required

Name of the S3 bucket to which to write data.

string

Optional

Name of the AWS region in which the target bucket resides. If you host your stream processing instance in an AWS region, this parameter defaults to that region. Otherwise, it defaults to the AWS region nearest your stream processing instance host region.

string | expression

Required

Prefix of the key of objects written to the S3 bucket. Must either be a literal prefix string or an expression that evaluates to a string.

document

Optional

Document containing additional parameters that override various default values.

document

Optional

Document containing additional parameters governing write behavior. These parameters trigger write behavior according to which threshold is met first.

For example, if the ingested documents reach the config.writeOptions.count threshold without reaching the config.writeOptions.interval threshold, the stream processor still emits these documents to S3 according to the config.writeOptions.count threshold.

integer

Optional

Number of documents to group into each file written to S3.

integer

Optional

Specifies the minimum number of bytes that must accumulate before a file is written to S3. The byte count is determined by the size of the BSON documents ingested by the pipeline, not the size of the final output file.

document

Optional

Specifies a timer for bulk writing documents as a combination of size and units.

Defaults to 1 minute. You can't set the size to 0 for any unit. The maximum interval is 7 days.

integer

Conditional

The number of units specified by writeOptions.interval.units after which the stream processor bulk writes documents to S3.

Defaults to 1. You can't set a size of 0. If you define writeOptions.interval, you must also define this parameter.

string

Conditional

The denomination of time in which to count the bulk write timer. This parameter supports the following values:

• ms

• second

• minute

• hour

• day

Defaults to minute. If you define writeOptions.interval, you must also define this parameter.

string

Optional

Delimiter between each entry in the emitted file.

Defaults to \n.

string

Optional

Specifies the output format of the JSON written to S3. Must be one of the following values:

• "basicJson"

• "canonicalJson"

• "relaxedJson"

Defaults to "relaxedJson".

To learn more, see Basic JSON.

string

Optional

Date format for the date value. Valid values are:

• default - to use the default of the outputFormat.

• ISO8601 - to convert dates to strings in the ISO8601 format, which includes millisecond precision (YYYY-MM-DDTHH:mm:ss.sssZ).

For example, if you add the following record to the pipeline:

{ "flightTime" : ISODate('2025-01-10T20:17:38.387Z') }

then if $emit.config.defaultFormat is set to default, output looks similar to the following:

{ "flightTime" : {$date :"2025-01-10T20:17:38.387Z"}}

If $emit.config.defaultFormat is set to ISO8601, output looks similar to the following:

{ "flightTime" : "2025-01-10T20:17:38.387Z" }

string

Optional

Name of the compression algorithm to use. Must be one of the following values:

• "gzip"

• "snappy"

string

Conditional

Level of compression to apply to the emitted message. Supports values 1-9 inclusive; higher values mean more compression.

Defaults to 6.

This parameter is required for and limited to gzip. If you set config.compression to snappy, setting this parameter has no effect.

{ "binary": { "$binary": { "base64": "gf1UcxdHTJ2HQ/EGQrO7mQ==", "subType": "00" }}}

{ "binary": "gf1UcxdHTJ2HQ/EGQrO7mQ=="}

{ "date": { "$date": "2024-10-24T18:07:29.636Z"}}

{ "date": 1729625275856}

{ "decimal": { "$numberDecimal": "9.9" }}

{ "decimal": "9.9" }

{ "timestamp": { "$timestamp": { "t": 1729793249, "i": 1 }}}

{ "timestamp": 1729793249000}

{ "_id": { "$oid": "671a8ce1497407eff0e17cba" }}

{ "_id": "6717fcbba18c8a8f74b6d977" }

{ "negInf": { "$numberDouble": "-Infinity" }}

{ "negInf": "-Infinity" }

{ "posInf": { "$numberDouble": "Infinity" }}

{ "posInf": "Infinity" }

{ "regex": { "$regularExpression": { "pattern": "ab+c", "options": "i" }}}

{ "regex": { "pattern": "ab+c", "options": "i" }}

{ "uuid": { "$binary": { "base64": "Kat+fHk6RkuAmotUmsU7gA==", "subType": "04" }}}

{ "uuid": "420b7ade-811a-4698-aa64-c8347c719cf1"}