/ /

`$emit` 聚合阶段（流处理）

定义

$emit 阶段会指定连接注册表中要向其发送消息的连接。支持以下连接类型：

Apache Kafka代理
MongoDB 时间序列集合
AWS Kinesis数据流
AWS S3 桶
Azure Blob 存储
Google Cloud Pub/Sub

展示位置

$emit 必须是它所在的任何管道的最后一个阶段。每个管道只能使用一个$emit阶段。

语法

Apache Kafka 代理

要将处理后的数据写入Apache Kafka代理，请使用具有以下原型形式的 $emit管道阶段：

{
  "$emit": {
    "connectionName": "<registered-connection>",
    "topic": "<target-topic>" | <expression>,
    "schemaRegistry": {
      "connectionName": "<schema-registry-name>",
      "valueSchema": {
        type: "<schema-type>",
        schema: <schema-name>,
        options: {
          subjectNameStrategy: "<topic-name-strategy>",
          autoRegisterSchemas: true
        }
      }
    },
    "config": {
      "acks": <number-of-acknowledgements>,
      "compression_type": "<compression-type>",
      "dateFormat": "default" | "ISO8601",
      "headers": "<expression>",
      "key": "<key-string>" | { key-document },
      "keyFormat": "<serialization-type>",
      "outputFormat": "basicJson" | "canonicalJson" | "relaxedJson",
      "tombstoneWhen": <expression>
    }
  }
}

$emit 阶段采用包含以下字段的文档：

字段

类型

必要性

说明

connectionName

字符串

必需

从中提取数据的连接的名称，如连接注册表所示。

topic

string 或表达式

必需

要向其发出消息的Apache Kafka主题的名称。

schemaRegistry

文档

Optional

使用模式注册表来支持写入Avro 序列化源的文档。

要启用此功能，必须创建模式注册表连接。

schemaRegistry.connectionName

字符串

可选的

用于 Avro 反序列化的模式注册表连接的名称。

schemaRegistry.valueSchema

文档|表达式

可选的

定义序列化模式属性的文档，或计算结果的表达式。

schemaRegistry.valueSchema.type

字符串

可选的

要使用模式注册表的序列化类型。Atlas Stream Processing目前支持通过模式注册表连接进行 "avro" 序列化。

schemaRegistry.valueSchema.schema

文档

可选的

定义模式声明。的文档。

schemaRegistry.valueSchema.options

文档

Optional

为模式注册表连接定义可选配置参数的文档。

schemaRegistry.valueSchema.options.autoRegisterSchemas

布尔

Optional

在处理具有无法识别模式的文档时，切换确定是否自动注册模式。如果设置为 false，则具有无法识别模式的文档将发送到死信队列（DLQ）。

默认值为 true。

schemaRegistry.valueSchema.options.subjectNameStrategy

字符串

可选的

确定自动注册模式的主题名称的方法。必须是以下之一：

"TopicNameStrategy"：使用Kafka {topic} 名称作为主题名称。
"RecordNameStrategy"：使用 Avro记录名称作为主题名称。
"TopicRecordNameStrategy"：使用Kafka {topic} 名称和 Avro记录名称的组合作为主题名称。

默认为 "TopicNameStrategy"。只有当 schemaRegistry.valueSchema.options.autoRegisterSchemas设立为 true 时，才能设立此参数。

config

文档

Optional

包含可覆盖各种默认值的字段的文档。

config.acks

字符串

Optional

成功执行 $emit 操作所需的来自 Apache Kafka 集群的确认次数。

默认值为 "all"。Atlas Stream Processing 支持以下值：

"-1"
"0"
"1"
"all"

config.compression_type

字符串

Optional

用于制造者生成的所有数据的压缩类型。默认值为“无”（即，不压缩）。有效值为：

none
gzip
snappy
lz4
zstd

压缩可用于完整批次的数据，因此批处理的效率会影响压缩比；批处理越多，压缩效果越好。

config.dateFormat

字符串

Optional

日期值的日期格式。有效值为：

default - 使用 outputFormat 的默认值。
ISO8601 - 将日期转换为 ISO8601 格式的字符串，其中包含毫秒精度 (YYYY-MM-DDTHH:mm:ss.sssZ)。

例如：

请考虑以下输入。

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

如果将 $emit.config.dateFormat 设置为 default，输出将类似于以下内容：

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

如果将 $emit.config.dateFormat 设置为 ISO8601，输出将类似于以下内容：

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

config.headers

表达式(expression)

Optional

要添加到输出消息的标题。您必须将 config.headers 指定为聚合表达式或字段引用。您不能提供包含动态表达式的原始内联对象或原始数组字面量作为 config.headers 的值。

在运行时，该表达式的计算结果必须为完整的标头对象或 { k, v } 个文档的数组。

如果表达式的计算结果为对象，Atlas Stream Processing 将根据该对象中的每个键值对构造一个标头，其中键是标头名称，值是标头值。

如果表达式的计算结果为数组，则它必须采用键值对对象数组的形式。例如：

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

Atlas Stream Processing 从数组中的每个对象构造一个标头，其中键是标头名称，值是标头值。 Atlas Stream Processing 支持以下类型的标头值：

binData
string
object
int
long
double
null

To learn how to build headers from field values or stream metadata, see Dynamic Kafka Headers.

config.key

对象 | 字符串 | 表达式

Optional

求值为 Apache Kafka 消息键的表达式。

如果您指定 config.key，则必须指定 config.keyFormat。

config.keyFormat

字符串

可选的

用于序列化Apache Kafka关键数据的数据类型。必须是以下值之一：

"binData"
"string"
"json"
"int"
"long"
"double"

默认为 binData。如果您指定 config.key，则必须指定 config.keyFormat。如果文档的 config.key 未成功序列化为指定的数据类型，Atlas Stream Processing 会将其发送到死信队列。

config.outputFormat

字符串

Optional

向 Apache Kafka 发送消息时使用的 JSON 格式。必须是以下值之一：

"basicJson"
"canonicalJson"
"relaxedJson"

默认值为 "relaxedJson"。

To learn more, see Basic JSON.

config.tombstoneWhen

表达式(expression)

Optional

用于确定何时向Kafka发出 null 的表达式。该表达式的计算结果必须为布尔值 true 或 false。当给定文档的表达式计算结果为 true 时， Atlas Stream Processing会向Kafka接收器发出一个 null 代替。当表达式的计算结果为 false 时， Atlas Stream Processing在到达 $emit 阶段时按原样发出文档。

如果表达式无法计算出布尔值，或者无法计算， Atlas Stream Processing会将文档写入 DLQ。

如果您提供 $emit.config.key 和 $emit.config.keyFormat 值，则此设置可用于启用主题压实。如果您不提供这些值，当该表达式的计算结果为 true 时， Atlas Stream Processing仍会发出 null，但不会触发Kafka主题压实。

Atlas 时间序列集合

要将处理后的数据写入 Atlas 时间序列集合，请使用具有以下原型形式的$emit管道阶段：

{
  "$emit": {
    "connectionName": "<registered-connection>",
    "db": "<target-db>" | <expression>,
    "coll": "<target-coll>" | <expression>,
    "parallelism": <number>,
    "timeseries": {
      <options>
    }
  }
}

$emit 阶段采用包含以下字段的文档：

字段	类型	必要性	说明
`connectionName`	字符串	必需	从中提取数据的连接的名称，如连接注册表所示。
`db`	字符串 \|表达式	必需	解析为包含目标时间序列集合的Atlas数据库的名称或表达式。
`coll`	字符串 \|表达式	必需	解析为要写入的Atlas时间序列集合的名称或表达式。
`parallelism`	整型	Optional	时间序列接收器操作符符用于将数据写入时间序列集合的内部写入线程的数量。这通过将写入操作分配给多个线程来提高大容量工作负载的吞吐量。默认值为 `1`。可设置的最大值取决于 Stream Processing 器的层级。要了解更多信息，请参阅 Atlas Stream Processing 层级选择指南。每个流处理器都有一个由其层级确定的最大累积并行度值。流处理器的累积并行度计算如下： `parallelism total - parallelized stages` Where `parallelism total` is the sum of all `parallelism` values greater than 1 across the `$source`, $lookup, $merge, $emit, and $externalFunction stages, and `parallelized stages` is the number of these stages with `parallelism` values greater than `1`. 示例，如果您的 `$source` 阶段将 `parallelism` 值设置为 `4`，则 `$lookup` 阶段未设置 `parallelism` 值（因此默认为 `1`），并且 `$merge` 阶段将 `parallelism` 值为 `2`，则有两个 `parallelized stages`，流处理器的累积并行度计算为 `(4 + 2) - 2`。如果流处理器超过其层级的最大累积并行度， Atlas Stream Processing会引发错误，并建议您实现预期并行度层级所需的最低处理器层级。您必须将处理器扩展到更高层级，或降低阶段的并行度值，才能解决该错误。要学习，了解更多信息，请参阅Stream Processing。
`timeseries`	文档	Optional	Document defining the time series fields for the collection.

注意

The maximum size for documents within a time series collection is 4 MB. To learn more, see Time Series Collection Limitations.

重要

动态表达式要求

在 db 或 coll 字段中使用表达式时，您必须在 timeseries 配置中指定 timeField 选项。这样可以确保当目标集合不存在时， Atlas Stream Processing可以正确创建时间序列集合。

例如：

{
  "$emit": {
    "connectionName": "atlas1",
    "db": "$targetDatabase",
    "coll": "$targetCollection",
    "timeseries": {
      "timeField": "timestamp"
    }
  }
}

AWS Kinesis

Atlas Stream Processing支持创建与 AWS Kinesis Streams 的 Private Link 连接。要学习；了解更多信息，请参阅添加Kinesis Private Link 连接。

要将经过处理的数据写入 AWS Kinesis，请使用 $emit 管道阶段，并采用以下原型形式：

{
  "$emit": {
    "connectionName": "<registered-connection>",
    "stream": "<stream-name>",
    "region": "<aws-region>",
    "partitionKey": "<key>" | <field> | <expression>
    "config": {
      "outputFormat": "basicJson" | "canonicalJson" | "relaxedJson",
      "dateFormat": "default" | "ISO8601",
    }
  }
}

$emit 阶段采用包含以下字段的文档：

字段

类型

必要性

说明

connectionName

字符串

必需

从中提取数据的连接的名称，如连接注册表所示。

stream

字符串

必需

要连接的Kinesis数据流的名称。

region

字符串

Optional

Kinesis Data Stream 运行的区域。AWS 支持多个同名流，每个流都位于不同的区域。此参数允许 Atlas Stream Processing 区分此类流。

partitionKey

字符串 |表达式

必需

用于确定哪个 Kinesis 分片接收每个记录的密钥。必须是文字 string 或计算结果为 string 的表达式。

config

文档

Optional

包含可覆盖各种默认值的字段的文档。

config.outputFormat

字符串

Optional

向Kinesis发送消息时使用的JSON格式。必须是以下值之一：

"basicJson"
"canonicalJson"
"relaxedJson"

默认值为 "relaxedJson"。

To learn more, see Basic JSON.

config.dateFormat

字符串

Optional

日期值的日期格式。有效值为：

default - 使用 outputFormat 的默认值。
ISO8601 - 将日期转换为 ISO8601 格式的字符串，其中包含毫秒精度 (YYYY-MM-DDTHH:mm:ss.sssZ)。

例如：

请考虑以下输入。

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

如果将 $emit.config.dateFormat 设置为 default，输出将类似于以下内容：

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

如果将 $emit.config.dateFormat 设置为 ISO8601，输出将类似于以下内容：

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

AWS S3

要将处理后的数据写入 AWS S3 存储桶的接收器连接，请使用 $emit 管道阶段，并采用以下原型形式：

{
  "$emit": {
    "connectionName": "<registered-connection>",
    "bucket": "<target-bucket>" | <expression>,
    "region": "<target-region>",
    "path": "<key-prefix>" | <expression>,
    "parallelism": <number>,
    "partitionBy": <expression>,
    "config": {
      "writeOptions": {
        "count": <doc-count>,
        "bytes": <threshold>,
        "interval": {
          "size": <unit-count>,
          "unit": "<time-denomination>"
        }
      },
      "delimiter": "<delimiter>",
      "outputFormat": "basicJson" | "canonicalJson" | "relaxedJson",
      "dateFormat": "default" | "ISO8601",
      "compression": "gzip" | "snappy",
      "compressionLevel": <level>
    }
  }
}

$emit 阶段采用包含以下字段的文档：

字段

类型

必要性

说明

connectionName

字符串

必需

从中写入数据的连接的名称，如连接注册表所示。

bucket

字符串 |表达式

必需

要写入数据的 S3 存储桶的名称，指定为字面 string 或计算结果为 string 的表达式。

region

字符串

Optional

目标存储桶所在的 AWS 区域的名称。如果您在 AWS 区域托管Stream Processing 工作区，此参数将默认为该区域。否则，它将默认为离您的流处理工作区托管区域最近的 AWS 区域。

path

字符串 |表达式

必需

写入 S3 存储桶的对象键的前缀。必须是字面前缀字符串或计算结果为字符串的表达式。

parallelism

整型

Optional

S3 接收器操作符符用于将数据写入S3 的内部写入线程的数量。这通过将写入操作分配给多个线程来提高大容量工作负载的吞吐量。

默认值为 1。

可设置的最大值取决于 Stream Processing 器的层级。要了解更多信息，请参阅 Atlas Stream Processing 层级选择指南。

每个流处理器都有一个由其层级确定的最大累积并行度值。流处理器的累积并行度计算如下：

parallelism total - parallelized stages

Where parallelism total is the sum of all parallelism values greater than 1 across the $source, $lookup, $merge, $emit, and $externalFunction stages, and parallelized stages is the number of these stages with parallelism values greater than 1.

示例，如果您的 $source 阶段将 parallelism 值设置为 4，则 $lookup 阶段未设置 parallelism 值（因此默认为 1），并且 $merge 阶段将 parallelism 值为 2，则有两个 parallelized stages，流处理器的累积并行度计算为 (4 + 2) - 2。

如果流处理器超过其层级的最大累积并行度， Atlas Stream Processing会引发错误，并建议您实现预期并行度层级所需的最低处理器层级。您必须将处理器扩展到更高层级，或降低阶段的并行度值，才能解决该错误。要学习，了解更多信息，请参阅Stream Processing。

partitionBy

表达式(expression)

Optional

用于确定当 parallelism 大于 1 时如何跨写入线程对文档进行分区的表达式。计算表达式的哈希值决定了处理每个文档的线程。

如果 parallelism 大于 1 但未指定 partitionBy， Atlas Stream Processing将使用循环策略将文档分配给线程。

config

文档

Optional

包含附加参数的文档，这些参数会覆盖各种默认值。

config.writeOptions

文档

Optional

包含控制写入行为的附加参数的文档。这些参数会根据首先满足的阈值触发写入行为。

例如，如果摄入的文档达到 config.writeOptions.count 阈值但未达到 config.writeOptions.interval 阈值，流处理器仍会根据 config.writeOptions.count 阈值将这些文档发送到 S3。

config.writeOptions.count

整型

Optional

写入 S3 的每个文件中要分组的文档数量。

config.writeOptions.bytes

整型

Optional

指定在将文件写入 S3 之前必须累积的最小字节数。字节计数由管道摄取的 BSON 文档的大小决定，而不是最终输出文件的大小。

默认为 32 MB。

config.writeOptions.interval

文档

Optional

指定用于批量写入文档的计时器，作为 size 和 unit 的组合。

默认为 1 分钟。您无法将任何 unit 的 size 设置为 0。最大间隔为 7 天。

config.writeOptions.interval.size

整型

可选的

流处理器会在达到 writeOptions.interval.unit 指定的单位数后，将文档批量写入 S3。

默认为 1。您不能将 size 设置为 0。如果您定义 writeOptions.interval，则还必须定义此参数。

config.writeOptions.interval.unit

字符串

可选的

用于计数批量写入计时器的时间单位。该参数支持以下值：

ms
second
minute
hour
day

默认为 minute。如果您定义 writeOptions.interval，则还必须定义此参数。

config.delimiter

字符串

Optional

生成文件中每个条目之间的分隔符。

默认值为 \n。

config.outputFormat

字符串

Optional

指定写入 S3 的 JSON 输出格式。必须是以下值之一：

"basicJson"
"canonicalJson"
"relaxedJson"

默认为“relaxedJson”。

To learn more, see Basic JSON and Extended JSON.

config.dateFormat

字符串

Optional

日期值的日期格式。有效值为：

default - 使用 outputFormat 的默认值。
ISO8601 - 将日期转换为 ISO8601 格式的字符串，其中包含毫秒精度 (YYYY-MM-DDTHH:mm:ss.sssZ)。

例如，如果您将以下记录添加到管道中：

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

那么，如果将 $emit.config.dateFormat 设置为 default，输出将类似于以下内容：

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

如果将 $emit.config.dateFormat 设置为 ISO8601，输出将类似于以下内容：

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

config.compression

字符串

Optional

要使用的压缩算法名称。必须是以下值之一：

"none"
"gzip"
"snappy"

config.compressionLevel

整型

可选的

应用于所发出消息的压缩级别。支持值包括 1-9；值越高，压缩率越高。

默认值为 6。

此参数仅在 config.compression 设置为 gzip 时有效。将其与任何其他压缩类型一起设置会返回错误。

Azure Blob 存储

要将经过处理的数据写入 Azure Blob 存储，请使用 $emit 管道阶段，并采用以下原型形式：

{
  "$emit": {
    "connectionName": "<registered-connection>",
    "container": "<storage-container>",
    "path": "<prefix-key>" | <expression>,
    "parallelism": <int>,
    "partitionBy": "<key>" | <expression>,
    "config": {
      "dateFormat": "default" | "ISO8601",
      "compression": "none" | "gzip" | "snappy",
      "compressionLevel": [1-9],
      "outputFormat": "relaxedJson" | "canonicalJson" | "basicJson",
      "delimiter": "<delimiter>",
      "writeOptions": {
        "count": <int>,
        "interval": {
          "size": <int>,
          "unit": "ms" | "second" | "minute" | "hour" | "day"
        },
        "bytes": <int>
      }
    }
  }
}

$emit 阶段采用包含以下字段的文档：

字段

类型

必要性

说明

connectionName

字符串

必需

连接注册表中显示的要写入数据的连接的名称。

container

字符串

必需

要写入blob 的存储容器。

path

字符串 |表达式

必需

写入Azure Blob 存储的对象密钥的前缀。必须是 string 字面量或计算结果为 string 的表达式。

config

文档

Optional

包含附加参数的文档，这些参数会覆盖各种默认值。

config.dateFormat

字符串

Optional

日期值的日期格式。有效值为：

default - 使用 outputFormat 的默认值。
ISO8601 - 将日期转换为 ISO8601 格式的字符串，其中包含毫秒精度 (YYYY-MM-DDTHH:mm:ss.sssZ)。

例如，如果您将以下记录添加到管道中：

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

那么，如果将 $emit.config.dateFormat 设置为 default，输出将类似于以下内容：

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

如果将 $emit.config.dateFormat 设置为 ISO8601，输出将类似于以下内容：

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

config.compression

字符串

Optional

要使用的压缩算法名称。必须是以下值之一：

"none"
"gzip"
"snappy"

如果您没有为此字段指定值， Atlas Stream Processing将在不压缩的情况下写入数据。

config.compressionLevel

整型

可选的

应用于所发出消息的压缩级别。支持值包括 1-9；值越高，压缩率越高。

默认值为 6。

此参数仅在 config.compression 设置为 gzip 时有效。将其与任何其他压缩类型一起设置会返回错误。

config.outputFormat

字符串

Optional

指定写入Azure Blob 存储的JSON的输出格式。必须是以下值之一：

"basicJson"
"canonicalJson"
"relaxedJson"

默认值为 "relaxedJson"。

To learn more, see Basic JSON and Extended JSON.

config.delimiter

字符串

Optional

生成文件中每个条目之间的分隔符。

默认值为 \n。

config.writeOptions

文档

Optional

包含控制写入行为的附加参数的文档。这些参数会根据首先满足的阈值触发写入行为。

示例，如果摄入的文档达到 config.writeOptions.count 阈值但未达到 config.writeOptions.interval 阈值，流处理器仍会根据 config.writeOptions.count 阈值将这些文档发送到 Azure Blob 存储。

config.writeOptions.count

整型

Optional

要群组到写入Azure Blob 存储的每个文件中的文档数量。

config.writeOptions.bytes

整型

Optional

指定在将文件写入Azure Blob 存储之前必须累积的最小字节数。字节计数由管道摄取的BSON文档的大小决定，而不是最终输出文件的大小。

默认为 32 MB。

config.writeOptions.interval

文档

Optional

指定用于批量写入文档的计时器，作为 size 和 unit 的组合。

默认为 1 分钟。您无法将任何 unit 的 size 设置为 0。最大间隔为 7 天。

config.writeOptions.interval.size

整型

可选的

writeOptions.interval.unit 指定的单元数，超过此数后流处理器会将文档批量写入Azure Blob 存储。

默认为 1。您不能将 size 设置为 0。如果您定义 writeOptions.interval，则还必须定义此参数。

config.writeOptions.interval.unit

字符串

可选的

用于计数批量写入计时器的时间单位。该参数支持以下值：

ms
second
minute
hour
day

默认为 minute。如果您定义 writeOptions.interval，则还必须定义此参数。

parallelism

整型

Optional

Atlas Stream Processing用于将数据写入Azure Blob 存储的内部写入线程数。这通过将写入操作分配给多个线程来提高大容量工作负载的吞吐量。

默认为 1。最大值为 16。

partitionBy

字符串 |表达式

Optional

字符串或表达式，用于确定当 parallelism 大于 1 时， Atlas Stream Processing如何跨写入线程对文档进行分区。计算表达式的哈希值决定了处理每个文档的线程。

如果 parallelism 大于 1 但未指定 partitionBy， Atlas Stream Processing将使用循环策略将文档分配给线程。

Google Cloud Pub/Sub

要将处理后的数据写入Google Cloud Pub/Sub，请使用具有以下原型形式的 $emit管道阶段：

{
  "$emit": {
    "connectionName": "<registered-connection>",
    "topic": "<topic>" | <expression>,
    "projectId": "<project-id>",
    "region": "<pubsub-region>",
    "orderingKey": "<key>" | <expression>,
    "attributes": {
      "<key1>": "<value1>" | <expression>,
      . . .
      "<keyN>": "<valueN>" | <expression>
    },
    "config": {
      "dateFormat": "default" | "ISO8601",
      "outputFormat": "relaxedJson"| "canonicalJson" | "basicJson",
    }
  }
}

$emit 阶段采用包含以下字段的文档：

字段

类型

必要性

说明

connectionName

字符串

必需

连接注册表中显示的要写入数据的连接的名称。

projectId

字符串

必需

要发布到的项目的ID 。

topic

字符串 |表达式

必需

给定项目中要发布消息的主题的名称。必须是 string 或计算结果为 string 的表达式。

region

字符串

必需

向其发送请求的位置终结点。有关可用区域，请参阅 Pub/Sub 文档。

orderingKey

字符串 |表达式

Optional

键入已发布的数据以命令消息发布。如果您设立orderingKey， Atlas Stream Processing将确保订阅者接收器按照源发布消息的顺序接收包含该密钥的所有已发布消息。

为此字段使用 string 值以设立静态 orderingKey。对于动态 orderingKey，请使用计算结果为 string 的表达式。专门计算字段路径（Field Path）的表达式将 orderingKey 设置为文档中字段的值。

attributes

文档

Optional

定义要添加到消息中的属性的文档。属性采用键值对的形式，其中：

每个键都是字符串或字段路径（Field Path）表达式，用于确定Atlas Stream Processing添加到消息的键。
每个值要么是字符串，要么是计算结果为字符串的字段路径（Field Path）表达式。

config

文档

Optional

包含附加参数的文档，这些参数会覆盖各种默认值。

config.dateFormat

字符串

Optional

日期值的日期格式。有效值为：

default - 使用 outputFormat 的默认值。
ISO8601 - 将日期转换为 ISO8601 格式的字符串，其中包含毫秒精度 (YYYY-MM-DDTHH:mm:ss.sssZ)。

例如，如果您将以下记录添加到管道中：

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

那么，如果将 $emit.config.dateFormat 设置为 default，输出将类似于以下内容：

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

如果将 $emit.config.dateFormat 设置为 ISO8601，输出将类似于以下内容：

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

config.outputFormat

字符串

Optional

指定写入 Google Cloud Pub/Sub 的JSON的输出格式。必须是以下值之一：

"basicJson"
"canonicalJson"
"relaxedJson"

默认值为 "relaxedJson"。

To learn more, see Basic JSON and Extended JSON.

Basic JSON

为了简化消息的摄取，$emit 阶段支持将处理后的数据以 Basic JSON 格式写入接收端，该格式简化了扩展（canonicalJson）和宽松扩展（relaxedJson）JSON 格式。Basic JSON 不使用 MongoDB 扩展 JSON 包装器，因此不保留所有 BSON types。

您可以通过在 $emit 阶段中将 config.outputFormat 字段设置为 "basicJson" 来指定基本 JSON 格式。

下表展示了所有受影响字段的简化示例。

字段类型	relaxedJson	basicJson
二进制文件	`{ "binary": { "$binary": { "base64": "gf1UcxdHTJ2HQ/EGQrO7mQ==", "subType": "00" }}}`	`{ "binary": "gf1UcxdHTJ2HQ/EGQrO7mQ=="}`
Date	`{ "date": { "$date": "2024-10-24T18:07:29.636Z"}}`	`{ "date": 1729625275856}`
Decimal 数据类型	`{ "decimal": { "$numberDecimal": "9.9" }}`	`{ "decimal": "9.9" }`
时间戳	`{ "timestamp": { "$timestamp": { "t": 1729793249, "i": 1 }}}`	`{ "timestamp": 1729793249000}`
ObjectId	`{ "_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	`{ "uuid": { "$binary": { "base64": "Kat+fHk6RkuAmotUmsU7gA==", "subType": "04" }}}`	`{ "uuid": "420b7ade-811a-4698-aa64-c8347c719cf1"}`

行为

每个流处理器只能写入一个 Atlas 时间序列集合。如果您指定的集合不存在，Atlas 将使用您指定的时间序列字段创建该集合。您必须指定一个现有数据库。

Atlas Stream Processing doesn't support writing BSON documents larger than 16MB with the $emit stage to an AWS or Azure blob storage sink. You can increase this limit to up to 125MB if your stream processor's source isn't a MongoDB deployment, or if earlier pipeline stages such as $lookup increase the size of your documents.

Azure Blob Storage $emit 变体还支持写入Azure Datalake Storage v2 容器。

The GCP Pub/Sub $emit variant supports defining values through field path expressions for both the orderingKey field and the key-value pairs in the attributes field. On expression failure or if the expression doesn't resolve to a string, Atlas Stream Processing sends the affected message to the dead letter queue.

You can use a field or dynamic expression as the value of the topic field to enable your stream processor to write to different targets on a message-by-message basis. The expression must evaluate to a string.

例子

您有一个事务事件流，它会生成以下形式的消息：

{
  "customer": "Very Important Industries",
  "customerStatus": "VIP",
  "tenantId": 1,
  "transactionType": "subscription"
}
{
  "customer": "N. E. Buddy",
  "customerStatus": "employee",
  "tenantId": 5,
  "transactionType": "requisition"
}
{
  "customer": "Khan Traktor",
  "customerStatus": "contractor",
  "tenantId": 11,
  "transactionType": "billableHours"
}

要将其中每一个分类到不同的Apache Kafka主题中，您可以写入以下 $emit 阶段：

{
  "$emit": {
    "connectionName": "kafka1",
    "topic": "$customerStatus"
  }
}

此$emit阶段：

将Very Important Industries消息写入名为VIP的主题。
将N. E. Buddy消息写入名为employee的主题。
将Khan Traktor消息写入名为contractor的主题。

For more information on dynamic expressions, see expression operators.

如果您指定的主题尚不存在，则Apache Kafka会在收到第一条指向该主题的消息时自动创建该主题。

If you specify a topic with a dynamic expression, but Atlas Stream Processing cannot evaluate the expression for a given message, Atlas Stream Processing sends that message to the dead letter queue if configured and processes subsequent messages. If there is no dead letter queue configured, then Atlas Stream Processing skips the message completely and processes subsequent messages.

动态 Kafka 标头

要动态构建 Apache Kafka 标头，请在较早的管道阶段构建完整的标头对象，然后从 $emit.config.headers 引用该对象。当标头是传输元数据且不应出现在已发出消息的正文中时，请使用 $setStreamMeta：

{
  "$setStreamMeta": {
    "stream.kafka_headers": {
      "$arrayToObject": [
        {
          "k": "test_header",
          "v": "$myField"
        }
      ]
    }
  }
},
{
  "$emit": {
    "connectionName": "kafka",
    "topic": "my-topic",
    "config": {
      "headers": {
        "$meta": "stream.kafka_headers"
      }
    }
  }
}

示例

要运行本节中的示例，请创建一个流处理工作区并添加与Atlas Stream Processing工作区的连接。要学习；了解操作方法，请参阅开始使用Atlas Stream Processing和添加Atlas Stream Processing连接。

写入Kafka主题

A streaming data source generates detailed weather reports from various locations, conformant to the schema of the Sample Weather Dataset. The following aggregation has three stages:

$source 阶段与 Apache Kafka 代理建立连接，在名为 my_weatherdata 的主题中收集这些报告，从而在将每条记录引入后续聚合阶段时将其公开。此阶段还会覆盖其投影的时间戳字段的名称，将其设置为 ingestionTime。
The $match stage excludes documents that have an airTemperature.value of greater than or equal to 30.0 and passes along the documents with an airTemperature.value less than 30.0 to the next stage.
The $addFields stage enriches the stream with metadata.
$emit 阶段会通过 weatherStreamOutput Kafka 代理连接将输出写入名为 stream 的主题。

{
  "$source": {
    "connectionName": "sample_weatherdata",
    "topic": "my_weatherdata",
    "tsFieldName": "ingestionTime"
  }
},
{
  "$match": {
    "airTemperature.value": {
      "$lt": 30
    }
  }
},
{
  "$addFields": {
    "processorMetadata": {
      "$meta": "stream"
    }
  }
},
{
  "$emit": {
    "connectionName": "weatherStreamOutput",
    "topic": "stream"
  }
}

stream 主题中的文档采用以下形式：

{
  "st": "x+34700+119500",
  "position": {
    "type": "Point",
    "coordinates": [122.8, 116.1]
  },
  "elevation": 9999,
  "callLetters": "6ZCM",
  "qualityControlProcess": "V020",
  "dataSource": "4",
  "type": "SAO",
  "airTemperature": {
    "value": 6.7,
    "quality": "9"
  },
  "dewPoint": {
    "value": 14.1,
    "quality": "1"
  },
  "pressure": {
    "value": 1022.2,
    "quality": "1"
  },
  "wind": {
    "direction": {
      "angle": 200,
      "quality": "9"
    },
    "type": "C",
    "speed": {
      "rate": 35,
      "quality": "1"
    }
  },
  "visibility": {
    "distance": {
      "value": 700,
      "quality": "1"
    },
    "variability": {
      "value": "N",
      "quality": "1"
    }
  },
  "skyCondition": {
    "ceilingHeight": {
      "value": 1800,
      "quality": "9",
      "determination": "9"
    },
    "cavok": "N"
  },
  "sections": ["AA1", "AG1", "UG1", "SA1", "MW1"],
  "precipitationEstimatedObservation": {
    "discrepancy": "0",
    "estimatedWaterDepth": 999
  },
  "atmosphericPressureChange": {
    "tendency": {
      "code": "4",
      "quality": "1"
    },
    "quantity3Hours": {
      "value": 3.8,
      "quality": "1"
    },
    "quantity24Hours": {
      "value": 99.9,
      "quality": "9"
    }
  },
  "seaSurfaceTemperature": {
    "value": 9.7,
    "quality": "9"
  },
  "waveMeasurement": {
    "method": "M",
    "waves": {
      "period": 8,
      "height": 3,
      "quality": "9"
    },
    "seaState": {
      "code": "00",
      "quality": "9"
    }
  },
  "pastWeatherObservationManual": {
    "atmosphericCondition": {
      "value": "6",
      "quality": "1"
    },
    "period": {
      "value": 3,
      "quality": "1"
    }
  },
  "skyConditionObservation": {
    "totalCoverage": {
      "value": "02",
      "opaque": "99",
      "quality": "9"
    },
    "lowestCloudCoverage": {
      "value": "00",
      "quality": "9"
    },
    "lowCloudGenus": {
      "value": "00",
      "quality": "1"
    },
    "lowestCloudBaseHeight": {
      "value": 1750,
      "quality": "1"
    },
    "midCloudGenus": {
      "value": "99",
      "quality": "1"
    },
    "highCloudGenus": {
      "value": "00",
      "quality": "1"
    }
  },
  "presentWeatherObservationManual": {
    "condition": "52",
    "quality": "1"
  },
  "atmosphericPressureObservation": {
    "altimeterSetting": {
      "value": 1015.9,
      "quality": "9"
    },
    "stationPressure": {
      "value": 1026,
      "quality": "1"
    }
  },
  "skyCoverLayer": {
    "coverage": {
      "value": "08",
      "quality": "1"
    },
    "baseHeight": {
      "value": 2700,
      "quality": "9"
    },
    "cloudType": {
      "value": "99",
      "quality": "9"
    }
  },
  "liquidPrecipitation": {
    "period": 12,
    "depth": 20,
    "condition": "9",
    "quality": "9"
  },
  "extremeAirTemperature": {
    "period": 99.9,
    "code": "N",
    "value": -30.4,
    "quantity": "1"
  },
  "ingestionTime": {
    "$date": "2024-09-26T17:34:41.843Z"
  }
}

注意

以上是一个有代表性的示例。流数据不是静态的，每个用户看到的都是不同的文档。

写入时间序列集合

以下聚合从示例流媒体数据源sample_stream_solar 读取报告并将其写入时间序列集合。此聚合分为三个阶段：

$source 阶段连接到 sample_stream_solar 源。
$set 阶段使用 $dateFromString 将每个报告的 timestamp字段从字符串转换为日期，以便时间序列集合可以将其用作 timeField。
$emit 阶段通过 solar-cluster-connection 连接将每个报告写入 solarDb数据库中的 solar_ts时间序列集合。 metaField 选项设立为 device_id，因此来自同一设备的报告会一起存储在存储桶中。如果 solar_ts集合不存在， Atlas Stream Processing会将其创建为时间序列集合。

{
  "$source": {
    "connectionName": "sample_stream_solar"
  }
},
{
  "$set": {
    "timestamp": {
      "$dateFromString": { "dateString": "$timestamp" }
    }
  }
},
{
  "$emit": {
    "connectionName": "solar-cluster-connection",
    "db": "solarDb",
    "coll": "solar_ts",
    "timeseries": {
      "timeField": "timestamp",
      "metaField": "device_id"
    }
  }
}

注意

sample_stream_solar 源是仅限测试的连接，每秒都会生成示例文档以进行快速原型设计。

solar_ts集合中的文档采用以下形式：

{
  "timestamp": {
    "$date": "2026-10-22T15:07:49.299Z"
  },
  "device_id": "device_0",
  "max_watts": 450,
  "_id": {
    "$oid": "6ada26c5ad54068b03089593"
  },
  "obs": {
    "watts": 135,
    "temp": 5
  },
  "event_type": 0,
  "group_id": 8
}