$emit 聚合阶段（流处理）

字符串

必需

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

字符串或表达式

必需

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

文档

Optional

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

int

Optional

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

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

• -1

• 0

• 1

• all

字符串

Optional

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

• none

• gzip

• snappy

• lz4

• zstd

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

字符串

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" }

表达式(expression)

Optional

要添加到输出消息的标题。该表达式的计算结果必须为对象或数组。

如果表达式的计算结果为对象，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

对象 | 字符串 | 表达式

Optional

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

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

字符串

可选的

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

• "binData"

• "string"

• "json"

• "int"

• "long"

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

字符串

Optional

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

• "relaxedJson"

• "canonicalJson"

默认值为 "relaxedJson"。

表达式(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主题压实。

字符串

必需

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

字符串

必需

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

字符串

Optional

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

文档

Optional

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

字符串

Optional

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

• "relaxedJson"

• "canonicalJson"

• `"basicJson"

默认值为 "relaxedJson"。

字符串

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" }

字符串

必需

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

字符串

必需

要写入数据的 S3 存储桶的名称。

字符串

Optional

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

字符串 |表达式

必需

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

文档

Optional

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

文档

Optional

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

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

整型

Optional

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

整型

Optional

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

文档

Optional

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

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

整型

可选的

writeOptions.interval.units 指定的单位数，之后流处理器将文档批量写入 S3。

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

字符串

可选的

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

• ms

• second

• minute

• hour

• day

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

字符串

Optional

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

默认值为 \n。

字符串

Optional

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

• "basicJson"

• "canonicalJson"

• "relaxedJson"

默认为“relaxedJson”。

要了解更多信息，请参阅 JSON。

字符串

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" }

字符串

Optional

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

• "gzip"

• "snappy"

字符串

可选的

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

默认值为 6。

此参数为必填项，且仅限用于 gzip。如果将 config.compression 设置为 snappy，设置此参数将没有效果。

{ "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"}