/ /

`$source` 阶段（流处理）

定义

$source

$source 阶段在连接注册表中指定要从中流式传输数据的连接。支持以下连接类型：

Apache Kafka 代理
MongoDB collection change stream
MongoDB database change stream
MongoDB 集群变更流
AWS Kinesis数据流
文档数组
Cron安排

语法

Apache Kafka 代理

为了操作来自Apache Kafka代理的流媒体数据， $source阶段具有以下原型形式：

{
  "$source": {
    "connectionName": "<registered-connection>",
    "topic" : ["<source-topic>", ...],
    "timeField": {
      $toDate | $dateFromString: <expression>
    },
    "partitionIdleTimeout": {
      "size": <duration-number>,
      "unit": "<duration-unit>"
    },
    "schemaRegistry": {
      "connectionName": "<schema-registry-name>",
    },
    "config": {
      "auto_offset_reset": "<start-event>",
      "group_id": "<group-id>",
      "keyFormat": "<deserialization-type>",
      "keyFormatError": "<error-handling>"
    },
  }
}

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

字段

类型

必要性

说明

connectionName

字符串

必需

用于标识连接注册表中要从中提取数据的连接的标签。

topic

字符串或字符串数组

必需

从中流式传输消息的一个或多个Apache Kafka主题的名称。如果要流多个主题的消息，请在数组中指定这些主题。

timeField

文档

Optional

为传入消息定义权威时间戳的文档。

如果使用timeField ，则必须将其定义为以下之一：

将源消息字段作为参数的表达式：$toDate。
将源消息字段作为参数的表达式：$dateFromString。

如果您未声明timeField ，Atlas Stream Processing 会根据源提供的消息时间戳创建一个时间戳。

partitionIdleTimeout

文档

Optional

指定分区在水印计算中被忽略之前允许其空闲的时间长度的文档。

默认情况下，此字段为禁用。要处理因空闲而无进展的分区，请为此字段设置一个值。

partitionIdleTimeout.size

整型

Optional

指定分区空闲超时持续时间的数字。

partitionIdleTimeout.unit

字符串

Optional

分区空闲超时持续时间的时间单位。

unit的值可以是以下值之一：

"ms" （毫秒）
"second"
"minute"
"hour"
"day"

schemaRegistry

文档

Optional

支持使用模式注册表来支持从Avro 序列化源中读取数据的文档。

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

schemaRegistry.connectionName

字符串

可选的

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

config

文档

Optional

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

config.auto_offset_reset

字符串

Optional

指定从Apache Kafka源主题中的哪个事件开始摄取。auto_offset_reset 采用以下值：

end、 latest或largest ：在初始化聚合时从主题中的最新事件开始摄取。
earliest、 beginning或smallest ：从主题中最早的事件开始摄取。

默认值为 latest。

config.group_id

字符串

Optional

Kafka 消费者组的 ID 用于与流处理器关联。如果省略，Atlas Stream Processing 会将 Stream Processing 工作区关联为以下格式的自动生成 ID：

asp-${streamProcessorId}-consumer

Atlas Stream Processing会自动为所有持久流处理器生成此参数的值。对于使用 sp 定义的临时流处理器。进程()，仅当您手动定义该参数时才会设立该参数。

config.enable_auto_commit

布尔

可选的

用于确定Kafka代理分区偏移提交策略的标志。Atlas Stream Processing支持两种提交策略：

如果将此参数设立为 true，则每次 $source 阶段将数据传递给下一个操作符时， Atlas Stream Processing都会提交偏移量。
如果将此参数设立为 false，流处理器会在Atlas Stream Processing获取检查点时提交分区偏移量。

对于使用 sp falsegroup_id定义的临时流处理器。进程()，此参数默认为，除非您设立了。否则默认为true 。

有关使用Kafka作为 $source 时偏移的更多信息，请参阅Kafka源和消费者组偏移。

config.keyFormat

字符串

Optional

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

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

默认值为 binData。

config.keyFormatError

字符串

Optional

如何处理在反序列化 Apache Kafka 密钥数据时遇到的错误。必须是以下值之一：

dlq，将文档写入您的死信队列。
passThrough，将文档发送到缺少关键数据的下一阶段。

注意

Atlas Stream Processing 要求源数据流中的文档在 json 或 ejson有效。Atlas Stream Processing 会将不满足此要求的文档设置为死信队列（如果您已配置）。

MongoDB 集合变更流

Atlas 集合变更流允许应用程序访问针对单个集合的实时数据更改。要学习如何对集合打开变更流，请参阅变更流。

使用变更流 $source 时，请为源集群配置至少 24 小时的 oplog 窗口。

为读取变更流，Atlas Stream Processing 会扫描 oplog 集合。因此，您可能会在日志中看到 COLLSCAN 警告。这些警告表示正常行为，并不表示错误。

如果您将 config.fullDocument 或 config.fullDocumentBeforeChange 配置为 required，请在要捕获的任何写入操作发生前，为每个集合启用 changeStreamPreAndPostImages。如果由于您在写入发生时尚未启用该功能，或后像已过期，导致某个事件的后像不可用，流处理器将失败。如需了解如何启用前像和后像，请参阅带文档前像和后像的变更流。

重要

仅当oplog集合仍包含其恢复令牌标识的操作时，您才能恢复变更流。为确保您的应用程序可以恢复其变更流，请将最小 Oplog 窗口配置为比您预期的最长中断更长的时间。

为了操作来自 Atlas collection 的流媒体数据， $source阶段具有以下原型形式：

{
  "$source": {
    "connectionName": "<registered-connection>",
    "timeField": {
      $toDate | $dateFromString: <expression>
    },
    "db" : "<source-db>",
    "coll" : ["<source-coll>",...],
    "initialSync": {
      "enable": <boolean>,
      "parallelism": <integer>
    },
    "readPreference": "<read-preference>",
    "readPreferenceTags": [
      {"<key>": "<value>"},
      . . .
    ],
    "config": {
      "startAfter": <start-token> | "startAtOperationTime": <timestamp>,
      "fullDocument": "<full-doc-condition>",
      "fullDocumentOnly": <boolean>,
      "fullDocumentBeforeChange": "<before-change-condition>",
      "pipeline": [{
        "<aggregation-stage>" : {
          <stage-input>,
          . . .
        },
        . . .
      }],
      "maxAwaitTimeMS": <time-ms>,
    }
  }
}

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

字段	类型	必要性	说明
`connectionName`	字符串	可选的	用于标识连接注册表中要从中提取数据的连接的标签。
`timeField`	文档	Optional	为传入消息定义权威时间戳的文档。如果使用`timeField` ，则必须将其定义为以下之一：将源消息字段作为参数的`$toDate`表达式将源消息字段作为参数的`$dateFromString`表达式。如果您未声明`timeField` ，Atlas Stream Processing 会根据源提供的消息时间戳创建一个时间戳。
`db`	字符串	必需	在`connectionName`指定的 Atlas 实例上托管的 MongoDB database 的名称。此数据库的change stream充当流媒体数据源。
`coll`	字符串或字符串数组	必需	在 `connectionName` 指定的Atlas实例上托管的一个或多个MongoDB集合的名称。这些集合的变更流流媒体数据源。如果省略此字段，则流处理器源自MongoDB数据库变更流。
`initialSync`	文档	Optional	包含 `initialSync` 配置参数的文档。 Atlas Stream Processing `initialSync` 允许您摄取Atlas集合中预先存在的文档，就像它们是插入 `changeEvent` 文档一样。如果启用`initialSync`，当您启动流处理器时，它会首先摄取并处理集合中的所有现有文档，然后再摄取和进程新传入的 `changeEvent` 文档。 `initialSync` 一旦完成，就不再重复。如果`coll` 命名了多个集合，则Atlas Stream Processing会同步列表中的每个集合。要学习；了解多集合同步顺序、层级限制和示例，请参阅配置多集合初始同步。 Atlas Stream Processing修复了流处理器启动时要同步的集合设立。它不会同步您在该点之后创建的集合，如果同步任何一个集合失败，整个流处理器都会失败。如果启用`initialSync`，则无法在管道中使用 `$hoppingWindow`、`$sessionWindow` 或 `$tumblingWindow` 阶段。重要提示：请参阅限制以了解`initialSync` 注意事项和约束。
`initialSync.enable`	布尔	可选的	确定是否启用`initialSync`。如果声明 `initialSync`字段，则必须设立此字段。
`initialSync.parallelism`	整型	Optional	确定进程`initialSync` 操作的并行度级别。如果不指定值，则默认为 `1`。如果 `coll` 指定了多个集合，则此值适用于整个集合列表，而不是分别适用于每个集合。 `initialSync` 将此值应用于具有任何受支持的 `_id` 类型的集合，包括Atlas Stream Processing无法以可预测顺序读取的类型。可设置的最大值取决于 Stream Processing 器的层级。要了解更多信息，请参阅 Atlas Stream Processing 层级选择指南。每个流处理器都有一个由其层级确定的最大累积并行度值。流处理器的累积并行度计算如下： `parallelism total - parallelized stages` 其中，`parallelism total` 是 `$source`、$lookup、$merge、$emit 和 $externalFunction 阶段中所有大于 1 的 `parallelism` 值的总和，`parallelized stages` 是 `parallelism` 值大于 `1`的这些阶段的数量。示例，如果您的 `$source` 阶段将 `parallelism` 值设置为 `4`，则 `$lookup` 阶段未设置 `parallelism` 值（因此默认为 `1`），并且 `$merge` 阶段将 `parallelism` 值为 `2`，则有两个 `parallelized stages`，流处理器的累积并行度计算为 `(4 + 2) - 2`。如果流处理器超过其层级的最大累积并行度， Atlas Stream Processing会引发错误，并建议您实现预期并行度层级所需的最低处理器层级。您必须将处理器扩展到更高层级，或降低阶段的并行度值，才能解决该错误。要学习，了解更多信息，请参阅Stream Processing。
`readPreference`	字符串	Optional	读取偏好（read preference）变更流和`initialSync` 操作的读取偏好。默认值为 `primary`。
`readPreferenceTags`	阵列	Optional	读取读取偏好（read preference）标签用于变更流和`initialSync` 操作。
`config`	文档	Optional	包含可覆盖各种默认值的字段的文档。
`config.startAfter`	token	可选的	源开始报告的变更事件。这采用 resume token 的形式。您只能使用`config.startAfter`或`config.startAtOperationTime`之一。
`config.startAtOperationTime`	时间戳日期	可选的	源应开始报告的操作时间。您只能使用`config.startAfter`或`config.startAtOperationTime`之一。接受 MongoDB 扩展 JSON `$date` 或 `$timestamp` 值。
`config.fullDocument`	字符串	可选的	用于控制变更流源是应返回完整文档还是仅在发生更新时返回变更的设置。必须是以下之一： `default` ：不为 `update` 操作返回完整文档。 `updateLookup` ：除了更新操作所做的更改之外，还返回完整文档的当前多数提交版本。 `required` ：必须返回完整的文档。如果没有完整文档，流处理器将失败。这不适应用删除操作，如果完整文档不可用，删除操作仍会创建删除事件。 `whenAvailable` ：只要有完整文档可用，就返回完整文档，否则返回更改。要将 `required` 或 `whenAvailable` 与集合变更流一起使用，您必须在该集合上启用变更流前像和后像。
`config.fullDocumentOnly`	布尔	可选的	用于控制change stream源是返回包括所有元数据的整个事件文档，还是仅返回`fullDocument`内容的设置。如果设置为`true` ，源仅返回`fullDocument`的内容。如果 `fullDocument`设立为 `required` 或 `whenAvailable`，则必须在该集合上启用变更流前图像和后图像。
`config.fullDocumentBeforeChange`	字符串	Optional	指定change stream源是否应在输出中包含处于原始“更改之前”状态的文档。必须是以下之一： `off` ：省略`fullDocumentBeforeChange`字段。 `required` ：必须以更改前的状态返回完整文档。如果处于更改前状态的完整文档不可用，则流处理器将失败。 `whenAvailable` ：只要有可用的文档，就返回处于更改前状态的完整文档，否则忽略`fullDocumentBeforeChange`字段。如果没有为`fullDocumentBeforeChange`指定值，则默认为`off` 。要将此字段与集合变更流一起使用，您必须在该集合上启用变更流前像和后像。
`config.pipeline`	文档	Optional	指定一个聚合管道，用于传递变更流输出进行进一步处理之前对其进行过滤。该管道必须符合修改变更流输出中描述的参数。重要提示：每个变更事件都包括 `wallTime` 和 `clusterTime` 字段。`$source` 之后的 Atlas Stream Processing 阶段期望接收到处理器摄取的这些字段。为确保正确处理变更流数据，请勿修改 `$source.config.pipeline` 中的这些字段。
`config.maxAwaitTimeMS`	整型	Optional	在返回空批次之前，等待新数据更改以报告给变更流游标的最长时间（以毫秒为单位）。默认值为 `1000`。

MongoDB database change stream

Atlas 数据库变更流允许应用程序访问针对单个数据库的实时数据更改。要学习如何针对数据库打开变更流，请参阅变更流。

使用变更流 $source 时，请为源集群配置至少 24 小时的 oplog 窗口。

为读取变更流，Atlas Stream Processing 会扫描 oplog 集合。因此，您可能会在日志中看到 COLLSCAN 警告。这些警告表示正常行为，并不表示错误。

重要

为了操作来自 Atlas 数据库变更流的流数据， $source阶段具有以下原型形式：

{
  "$source": {
    "connectionName": "<registered-connection>",
    "timeField": {
      $toDate | $dateFromString: <expression>
    },
    "db" : "<source-db>",
    "readPreference": "<read-preference>",
    "readPreferenceTags": [
      {"<key>": "<value>"},
      . . .
    ],
    "config": {
      "startAfter": <start-token> | "startAtOperationTime": <timestamp>,
      "fullDocument": "<full-doc-condition>",
      "fullDocumentOnly": <boolean>,
      "fullDocumentBeforeChange": "<before-change-condition>",
      "pipeline": [{
        "<aggregation-stage>" : {
          <stage-input>,
          . . .
        },
        . . .
      }]
    },
  }
}

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

字段	类型	必要性	说明
`connectionName`	字符串	可选的	用于标识连接注册表中要从中提取数据的连接的标签。
`timeField`	文档	Optional	为传入消息定义权威时间戳的文档。如果使用`timeField` ，则必须将其定义为以下之一：将源消息字段作为参数的`$toDate`表达式将源消息字段作为参数的`$dateFromString`表达式。如果您未声明`timeField` ，Atlas Stream Processing 会根据源提供的消息时间戳创建一个时间戳。
`db`	字符串	必需	在`connectionName`指定的 Atlas 实例上托管的 MongoDB database 的名称。此数据库的change stream充当流媒体数据源。
`readPreference`	字符串	Optional	变更流操作的读取偏好（read preference）。默认值为 `primary`。
`readPreferenceTags`	阵列	Optional	读取偏好（read preference）标签 for 变更流操作。
`config`	文档	Optional	包含可覆盖各种默认值的字段的文档。
`config.startAfter`	token	可选的	源开始报告的变更事件。这采用 resume token 的形式。您只能使用`config.startAfter`或`config.startAtOperationTime`之一。
`config.startAtOperationTime`	时间戳日期	可选的	源应开始报告的操作时间。您只能使用`config.startAfter`或`config.startAtOperationTime`之一。接受 MongoDB 扩展 JSON `$date` 或 `$timestamp` 值。
`config.fullDocument`	字符串	可选的	用于控制变更流源是应返回完整文档还是仅在发生更新时返回变更的设置。必须是以下之一： `default` : 使用服务器默认行为。: 不为 `update` 操作返回完整文档。 `updateLookup` ：除了更新操作所做的更改之外，还返回完整文档的当前多数提交版本。 `required` ：必须返回完整的文档。如果没有完整文档，流处理器将失败。这不适应用删除操作，如果完整文档不可用，删除操作仍会创建删除事件。 `whenAvailable` ：只要有完整文档可用，就返回完整文档，否则返回更改。如果没有为`fullDocument`指定值，则默认为`default` 。要将 `required` 或 `whenAvailable` 与数据库变更流一起使用，您必须在该数据库中的每个集合上启用变更流前像和后像。
`config.fullDocumentOnly`	布尔	可选的	用于控制change stream源是返回包括所有元数据的整个事件文档，还是仅返回`fullDocument`内容的设置。如果设置为`true` ，源仅返回`fullDocument`的内容。如果 `fullDocument`设立为 `required` 或 `whenAvailable`，则必须对该数据库中的每个集合启用变更流前像和后像。
`config.fullDocumentBeforeChange`	字符串	Optional	指定change stream源是否应在输出中包含处于原始“更改之前”状态的文档。必须是以下之一： `off` ：省略`fullDocumentBeforeChange`字段。 `required` ：必须以更改前的状态返回完整文档。如果处于更改前状态的完整文档不可用，则流处理器将失败。 `whenAvailable` ：只要有可用的文档，就返回处于更改前状态的完整文档，否则忽略`fullDocumentBeforeChange`字段。如果没有为`fullDocumentBeforeChange`指定值，则默认为`off` 。要将此字段与数据库变更流一起使用，您必须对该数据库中的每个集合启用变更流前像和后像。
`config.pipeline`	文档	Optional	指定一个聚合管道，用于过滤源点的变更流输出。该管道必须符合修改变更流输出中描述的参数。重要提示：每个变更事件都包括 `wallTime` 和 `clusterTime` 字段。`$source` 之后的 Atlas Stream Processing 阶段期望接收到处理器摄取的这些字段。为确保正确处理变更流数据，请勿修改 `$source.config.pipeline` 中的这些字段。
`config.maxAwaitTimeMS`	整型	Optional	在返回空批次之前，等待新数据更改以报告给变更流游标的最长时间（以毫秒为单位）。默认值为 `1000`。

MongoDB 集群范围变更流源

使用变更流 $source 时，请为源集群配置至少 24 小时的 oplog 窗口。

为读取变更流，Atlas Stream Processing 会扫描 oplog 集合。因此，您可能会在日志中看到 COLLSCAN 警告。这些警告表示正常行为，并不表示错误。

重要

要对来自整个 Atlas 集群变更流的流数据进行操作， $source 阶段具有以下原型形式：

{
  "$source": {
    "connectionName": "<registered-connection>",
    "timeField": {
      $toDate | $dateFromString: <expression>
    },
    "readPreference": "<read-preference>",
    "readPreferenceTags": [
      {"<key>": "<value>"},
      . . .
    ],
    "config": {
      "startAfter": <start-token> | "startAtOperationTime": <timestamp>,
      "fullDocument": "<full-doc-condition>",
      "fullDocumentOnly": <boolean>,
      "fullDocumentBeforeChange": "<before-change-condition>",
      "pipeline": [{
        "<aggregation-stage>" : {
          <stage-input>,
          . . .
        },
        . . .
      }]
    },
  }
}

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

字段	类型	必要性	说明
`connectionName`	字符串	可选的	用于标识连接注册表中要从中提取数据的连接的标签。
`timeField`	文档	Optional	为传入消息定义权威时间戳的文档。如果使用`timeField` ，则必须将其定义为以下之一：将源消息字段作为参数的`$toDate`表达式将源消息字段作为参数的`$dateFromString`表达式。如果您未声明`timeField` ，Atlas Stream Processing 会根据源提供的消息时间戳创建一个时间戳。
`readPreference`	字符串	Optional	变更流操作的读取偏好（read preference）。默认值为 `primary`。
`readPreferenceTags`	阵列	Optional	读取偏好（read preference）标签 for 变更流操作。
`config`	文档	Optional	包含可覆盖各种默认值的字段的文档。
`config.startAfter`	token	可选的	源开始报告的变更事件。这采用 resume token 的形式。您只能使用`config.startAfter`或`config.startAtOperationTime`之一。
`config.startAtOperationTime`	日期 \| 时间戳	可选的	源应开始报告的操作时间。您只能使用`config.startAfter`或`config.startAtOperationTime`之一。接受 MongoDB 扩展 JSON `$date` 或 `$timestamp` 值。
`config.fullDocument`	字符串	可选的	用于控制变更流源是应返回完整文档还是仅在发生更新时返回变更的设置。必须是以下之一： `default` : 使用服务器默认行为。: 不为 `update` 操作返回完整文档。 `updateLookup` ：除了更新操作所做的更改之外，还返回完整文档的当前多数提交版本。 `required` ：必须返回完整的文档。如果没有完整文档，流处理器将失败。这不适应用删除操作，如果完整文档不可用，删除操作仍会创建删除事件。 `whenAvailable` ：只要有完整文档可用，就返回完整文档，否则返回更改。如果没有为`fullDocument`指定值，则默认为`default` 。要将 `required` 或 `whenAvailable` 与集群变更流一起使用，您必须在该集群中的每个集合上启用变更流前图像和后图像。
`config.fullDocumentOnly`	布尔	可选的	用于控制change stream源是返回包括所有元数据的整个事件文档，还是仅返回`fullDocument`内容的设置。如果设置为`true` ，源仅返回`fullDocument`的内容。如果 `fullDocument`设立为 `required` 或 `whenAvailable`，则必须对该集群中的每个集合启用变更流前图像和后图像。
`config.fullDocumentBeforeChange`	字符串	Optional	指定change stream源是否应在输出中包含处于原始“更改之前”状态的文档。必须是以下之一： `off` ：省略`fullDocumentBeforeChange`字段。 `required` ：必须以更改前的状态返回完整文档。如果处于更改前状态的完整文档不可用，则流处理器将失败。 `whenAvailable` ：只要有可用的文档，就返回处于更改前状态的完整文档，否则忽略`fullDocumentBeforeChange`字段。如果没有为`fullDocumentBeforeChange`指定值，则默认为`off` 。要将此字段与数据库变更流一起使用，您必须对该数据库中的每个集合启用变更流前像和后像。
`config.pipeline`	文档	Optional	指定一个聚合管道，用于过滤源点的变更流输出。该管道必须符合修改变更流输出中描述的参数。请注意， Atlas Stream Processing希望从每个摄取的变更事件中接收 `wallTime` 和 `clusterTime` 字段。为确保正确处理变更流数据，请勿修改 `$source.config.pipeline` 中的这些字段。
`config.maxAwaitTimeMS`	整型	Optional	在返回空批次之前，等待新数据更改以报告给变更流游标的最长时间（以毫秒为单位）。默认值为 `1000`。

AWS Kinesis Data Stream

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

为了对来自 AWS Kinesis数据流的数据进行操作，$source 阶段具有以下原型形式：

{
  "$source": {
    "connectionName": "<registered-connection>",
    "stream": "<stream-name>",
    "region": "<aws-region>",
    "timeField": {
      $toDate | $dateFromString: <expression>
    },
    "tsFieldName": "<field-name>",
    "shardIdleTimeout": {
      "size": <duration-number>,
      "unit": "<duration-unit>"
    },
    "config": {
      "consumerARN": "<aws-arn>",
      "initialPosition": <initial-position>,
      reshardDetectionIntervalSecs: <interval>
    }
  }
}

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

字段	类型	必要性	说明
`connectionName`	字符串	必需	用于标识连接注册表中连接的标签，从中提取数据。
`config.consumerARN`	字符串	必需	ARN 与Kinesis使用方相对应的。您的使用者必须使用增强型扇出。
`stream`	字符串	必需	要从其中传输消息的AWS Kinesis数据流。
`region`	字符串	可选的	指定流所在的 AWS 区域。Kinesis支持在不同区域具有相同名称的多个数据流。如果您在同一连接内的两个或多个区域中对数据流使用相同的名称，则必须使用此字段来指定要使用的名称和区域的组合。
`timeField`	文档	Optional	为传入消息定义权威时间戳的文档。如果使用`timeField` ，则必须将其定义为以下之一：将源消息字段作为参数的`$toDate`表达式将源消息字段作为参数的`$dateFromString`表达式。如果您未声明`timeField` ，Atlas Stream Processing 会根据源提供的消息时间戳创建一个时间戳。
`tsFieldName`	字符串	Optional	投影文档中用于时间戳的字段名。使用此字段以覆盖默认时间戳字段名。
`shardIdleTimeout`	文档	Optional	指定分片在水印计算中被忽略之前允许分片的时间长度的文档。默认下，此字段处于禁用状态。要处理由于空闲而不向前移动的分片，请为此字段设立一个值。
`shardIdleTimeout.size`	文档	Optional	指定分片空闲超时持续时间的数字。
`shardIdleTimeout.unit`	文档	Optional	分片空闲超时持续时间的时间单位。 `unit`的值可以是以下值之一： `"ms"` （毫秒） `"second"` `"minute"` `"hour"` `"day"`
`config`	文档	Optional	包含可覆盖各种默认值的字段的文档。
`config.initialPosition`	字符串	Optional	开始摄取消息的 Kinesis 数据流历史中的位置。必须是以下之一： `"TRIM_HORIZON"`：从分片最早的消息开始摄取。 `"LATEST"`：从分片的最新消息开始摄取。 `"AT_TIMESTAMP"`: 从特定时间戳开开始摄取。需要 `config.atTimestamp`。 `"AT_SEQUENCE_NUMBER"`: 开始以特定序列号吞吐。 `"AFTER_SEQUENCE_NUMBER"`: 在特定序列号后开始摄取。默认值为 `"LATEST"`。
`config.atTimestamp`	日期	可选的	开始摄取消息的时间戳。当 `config.initialPosition` 为 `"AT_TIMESTAMP"` 时必需。
`config.reshardDetectionIntervalSecs`	整型	Optional	为了重新分片而检查流经Kinesis流的数据速率的时间间隔（以秒为单位）。默认为 `300` 秒。

文档数组

为了对文档数组进行操作， $source阶段具有以下原型形式：

{
  "$source": {
    "timeField": {
      $toDate | $dateFromString: <expression>
    },
    "documents" : [{source-doc},...] | <expression>
  }
}

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

字段	类型	必要性	说明
`timeField`	文档	Optional	为传入消息定义权威时间戳的文档。如果使用`timeField` ，则必须将其定义为以下之一：将源消息字段作为参数的`$toDate`表达式将源消息字段作为参数的`$dateFromString`表达式。如果您未声明`timeField` ，Atlas Stream Processing 会根据源提供的消息时间戳创建一个时间戳。
`documents`	阵列	可选的	用作流媒体数据源的文档数组。该字段的值可以是对象数组，也可以是计算结果为对象数组的表达式。使用`connectionName`字段时请勿使用此字段。

Cron 计划

要安排生成文档而不是从连接读取，$source 阶段具有以下原型形式：

{
  "$source": {
    "schedule": "<cron-expression>",
    "tsFieldName": "<timestamp-field-name>"
  }
}

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

字段	类型	必要性	说明
`schedule`	字符串	必需	六字段 cron表达式，用于确定阶段何时发出文档。这些字段依次为：秒、分、小时、月中的某一天、月份和星期几。 Atlas Stream Processing以 UTC 格式计算表达式的值。每个字段接受 `*` 作为所有值、单个值、范围（如 `1-5`）、步长（如 `0/15`）或逗号分隔的列表（如 `1,15,30`）。您可以使用名称 `JAN` 到 `DEC` 表示月份，使用 `SUN` 到 `SAT` 表示星期。请勿包含 `CRON_TZ` 前缀。要学习；了解Atlas Stream Processing如何解析限制月份和星期几的表达式，请参阅行为。
`tsFieldName`	字符串	Optional	该阶段将计划时间戳投影到的字段的名称。默认为 `_ts`。

行为

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

对于Kafka $source 阶段， Atlas Stream Processing会并行读取源主题内的多个分区。分区限制由处理器层级决定。如需学习；了解更多信息，请参阅流处理计费参考文档。

对于 cron $source 阶段，每个计划时间都会生成一个带有计划时间戳的空文档。使用管道中的后续阶段来填充文档。

当 schedule表达式同时限制一个月中的某一天和一周中的某一天时， Atlas Stream Processing仅在满足这两个字段的计划时间发出文档。这与 cron 实现不同，后者在任一字段匹配时发出文档。

如果具有 cron $source 阶段的流处理器重新启动，则在未运行时，它会为错过的每个计划时间发出一个文档。在您停止发送后，它不会在错过的计划时间内发出文档。

具有 cron $source 阶段的流处理器在计划时间之间连续运行， Atlas在整个时间段内在处理器层级对其进行计费。不频繁的安排不会降低处理器运行的费用。如需学习；了解更多信息，请参阅流处理计费参考文档。

示例

Kafka 示例

流数据源从不同位置生成详细的天气报告，符合示例天气数据集的模式。以下聚合分为三个阶段：

$source 阶段与Apache Kafka代理建立连接，在名为 my_weatherdata 的主题中收集这些报告，从而在将每条记录摄取到后续聚合阶段时将其公开。此阶段还会覆盖其投影的时间戳字段的名称，将其设置为 ingestionTime。
$match 阶段会排除 dewPoint.value 小于或等于 5.0 的文档，并将 dewPoint.value 大于 5.0 的文档传递到下一个阶段。
$merge 阶段将输出写入 sample_weatherstream 数据库中名为 stream 的 Atlas 集合。如果不存在此类数据库或集合，Atlas 会创建它们。

[{
  "$source": {
    "connectionName": "sample_weatherdata",
    "topic": "my_weatherdata"
  }
},
{
  "$match": { "dewPoint.value": { "$gt": 5 } }
},
{
  "$merge": {
    "into": {
      "connectionName": "weatherStreamOutput",
      "db": "sample_weatherstream",
      "coll": "stream"
    }
  }
}]

要查看生成的 sample_weatherstream.stream 集合中的文档，请连接到您的 Atlas 集群并运行以下命令：

db.getSiblingDB("sample_weatherstream").stream.find()

{
  _id: ObjectId('66ad2edfd4fcac13b1a28ce3'),
  airTemperature: { quality: '1', value: 27.7 },
  atmosphericPressureChange: {
    quantity24Hours: { quality: '9', value: 99.9 },
    quantity3Hours: { quality: '1' },
    tendency: { code: '1', quality: '1' }
  },
  atmosphericPressureObservation: {
    altimeterSetting: { quality: '1', value: 1015.9 },
    stationPressure: { quality: '1', value: 1021.9 }
  },
  callLetters: 'CGDS',
  dataSource: '4',
  dewPoint: { quality: '9', value: 25.7 },
  elevation: 9999,
  extremeAirTemperature: {
    code: 'N',
    period: 99.9,
    quantity: '9',
    value: -30.4
  },
  ingestionTime: ISODate('2024-08-02T19:09:18.071Z'),
  liquidPrecipitation: {
    condition: '9',
    depth: 160,
    period: 24,
    quality: '2'
  },
  pastWeatherObservationManual: {
    atmosphericCondition: { quality: '1', value: '8' },
    period: { quality: '9', value: 3 }
  },
  position: { coordinates: [ 153.3, 50.7 ], type: 'Point' },
  precipitationEstimatedObservation: {
    discrepancy: '4',
    estimatedWaterDepth: 4
  },
  presentWeatherObservationManual: { condition: '53', quality: '1' },
  pressure: { quality: '1', value: 1016.3 },
  qualityControlProcess: 'V020',
  seaSurfaceTemperature: { quality: '9', value: 27.6 },
  sections: [ 'AA2', 'SA1', 'MW1', 'AG1', 'GF1' ],
  skyCondition: {
    cavok: 'N',
    ceilingHeight: { determination: 'C', quality: '1', value: 6900 }
  },
  skyConditionObservation: {
    highCloudGenus: { quality: '1', value: '05' },
    lowCloudGenus: { quality: '9', value: '03' },
    lowestCloudBaseHeight: { quality: '9', value: 150 },
    lowestCloudCoverage: { quality: '1', value: '05' },
    midCloudGenus: { quality: '9', value: '08' },
    totalCoverage: { opaque: '99', quality: '1', value: '06' }
  },
  skyCoverLayer: {
    baseHeight: { quality: '9', value: 99999 },
    cloudType: { quality: '9', value: '05' },
    coverage: { quality: '1', value: '04' }
  },
  st: 'x+35700-027900',
  type: 'SAO',
  visibility: {
    distance: { quality: '1', value: 4000 },
    variability: { quality: '1', value: 'N' }
  },
  waveMeasurement: {
    method: 'I',
    seaState: { code: '99', quality: '9' },
    waves: { height: 99.9, period: 14, quality: '9' }
  },
  wind: {
    direction: { angle: 280, quality: '9' },
    speed: { quality: '1', rate: 30.3 },
    type: '9'
  }
}

注意

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

变更流示例

以下聚合会从 cluster0-collection 源引入数据，而该源会连接到已加载示例数据集的 Atlas 集群。要了解如何创建 Stream Processing 工作区并将 Atlas 集群连接添加到连接注册表，请参阅Atlas Stream Processing 入门。此聚合会运行两个阶段以打开变更流并记录对 sample_weatherdata 数据库中 data 集合的更改：

$source 阶段会连接到 cluster0-collection 源，并针对 sample_weatherdata 数据库中的 data 集合打开变更流。
$merge 阶段会将过滤后的变更流文档写入 sample_weatherdata 数据库中名为 data_changes 的 Atlas 集合。如果不存在此类集合，Atlas 则会进行创建。

[{
  "$source": {
    "connectionName": "cluster0-connection",
    "db": "sample_weatherdata",
    "coll": "data"
  }
},
{
  "$merge": {
    "into": {
      "connectionName": "cluster0-connection",
      "db": "sample_weatherdata",
      "coll": "data_changes"
    }
  }
}]

以下 mongosh 命令会删除 data 文档：

db.getSiblingDB("sample_weatherdata").data.deleteOne(
   { _id: ObjectId("5553a99ae4b02cf715120e4b") }
)

删除 data 文档后，流处理器会将此变更流事件文档写入 sample_weatherdata.data_changes 集合。要查看生成的 sample_weatherdata.data_changes 集合中的文档，请使用 mongosh 连接到您的 Atlas 集群并运行以下命令：

db.getSiblingDB("sample_weatherdata").data_changes.find()

[
  {
    _id: {
      _data: '8267A3D7A3000000012B042C0100296E5A1004800951B8EDE4430AB5C1B254BB3C96D6463C6F7065726174696F6E54797065003C64656C6574650046646F63756D656E744B65790046645F696400645553A99AE4B02CF715120E4B000004'
    },
    clusterTime: Timestamp({ t: 1738790819, i: 1 }),
    documentKey: { _id: ObjectId('5553a99ae4b02cf715120e4b') },
    ns: { db: 'sample_weatherdata', coll: 'data' },
    operationType: 'delete',
    wallTime: ISODate('2025-02-05T21:26:59.313Z')
  }
]

数据库变更流示例

以下聚合从 db-change-stream-connection 源摄取数据，该源连接到加载了 Sample Mflix Dataset Collections 数据集的Atlas 集群。要学习；了解如何创建流处理工作区以及如何将与Atlas 集群的连接添加到连接注册表中，请参阅Atlas Stream Processing入门。此聚合运行两个阶段，以打开针对 sample_mflix 源数据库的变更流，并在 sample_mflix_changes 接收器数据库中记录对 db_changes集合的更改：

$source 阶段连接到 db-change-stream-connection 源并打开针对 sample_mflix 源数据库的变更流。 config.startAtOperationTime字段设置源开始报告的时间。此示例将 startAt 值设置为从过去一分钟开始。
$merge 阶段将变更流文档写入 sample_mflix_changes 接收器数据库中名为 db_changes 的Atlas集合。

const startAt = new Date(Date.now() - 60 * 1000);
const pipeline = [
   {
      $source: {
         connectionName: "db-change-stream-connection",
         db: "sample_mflix",
         config: {
            startAtOperationTime: startAt
         }
      }
   },
   {
      $merge: {
         into: {
            connectionName: "db-change-stream-connection",
            db: "sample_mflix_changes",
            coll: "db_changes"
         }
      }
   }
];

对源集群运行以下命令，验证流处理器的行为。要进行连接，请参阅通过mongosh连接到集群。

在 sample_mflix 源数据库的movies集合中插入一个文档，在 comments集合中插入一个文档：

db.getSiblingDB("sample_mflix").movies.insertOne({
   title: "The Stream Processor",
   year: 2026
})
db.getSiblingDB("sample_mflix").comments.insertOne({
   name: "Ada Lovelace",
   text: "A fine film about data in motion."
})

插入文档后，流处理器会将每次插入的变更流事件文档写入 sample_mflix_changes.db_changes集合。对集群运行以下命令，查看生成的 sample_mflix_changes.db_changes集合中的文档：

db.getSiblingDB("sample_mflix_changes").db_changes.find(
   {},
   { _id: 0, clusterTime: 1, ns: 1, operationType: 1, fullDocument: 1 }
)

[
  {
    clusterTime: Timestamp({ t: 1787778766, i: 1 }),
    fullDocument: {
      _id: ObjectId('6a8f56ce73c5d4d0ee569a0d'),
      title: 'The Stream Processor',
      year: 2026
    },
    ns: { db: 'sample_mflix', coll: 'movies' },
    operationType: 'insert'
  },
  {
    clusterTime: Timestamp({ t: 1787778771, i: 1 }),
    fullDocument: {
      _id: ObjectId('6a8f56d373c5d4d0ee569a0e'),
      name: 'Ada Lovelace',
      text: 'A fine film about data in motion.'
    },
    ns: { db: 'sample_mflix', coll: 'comments' },
    operationType: 'insert'
  }
]

每个事件都包括时间戳、完整文档更改、operationType 和 ns字段。 ns字段命名源数据库和集合，以便您可以区分哪个集合产生了每个更改。

集群范围的变更流示例

以下聚合从 cluster-changestream-connection 源摄取数据，该源连接到加载了样本 Mflix 数据集集合数据集的Atlas 集群，并写入到 cluster-changestream-sink-connection 目标，该目标连接到单独的Atlas 集群。要学习；了解如何创建流处理工作区以及如何将与Atlas 集群的连接添加到连接注册表中，请参阅Atlas Stream Processing入门。此聚合运行两个阶段，以在源集群上打开集群范围的变更流，并在目标集群上的 cluster_changes数据库中记录对 events集合的更改。此 $source 阶段省略了 db 和 coll，因此它报告源集群上每个数据库和集合的更改，而不是单个集合或数据库的更改。

$source 阶段连接到 cluster-changestream-connection 源并打开针对整个源集群的变更流。 config.startAtOperationTime字段指定流处理器应开始报告在指定时间或之后发生的更改。
$merge 阶段将变更流文档写入目标集群上 cluster_changes数据库中名为 events 的Atlas集合。如果不存在此类数据库或集合， Atlas会进行创建。

[{
  "$source": {
    "connectionName": "cluster-changestream-connection",
    "config": {
      "startAtOperationTime": {"$date": "2024-08-19T18:00:00.000Z"}
    }
  }
},
{
  "$merge": {
    "into": {
      "connectionName": "cluster-changestream-sink-connection",
      "db": "cluster_changes",
      "coll": "events"
    }
  }
}]

以下 mongosh 命令将文档插入到源集群上 sample_mflix数据库的 movies集合中：

db.getSiblingDB("sample_mflix").movies.insertOne(
   { _id: ObjectId("66c1a1f1f1f1f1f1f1f1f1f1"), title: "Example Movie" }
)

插入文档后，流处理器会将变更流事件文档写入目标集群上的 cluster_changes.events集合。

要查看生成的 cluster_changes.events集合中的文档，请对目标集群运行以下命令。要进行连接，请参阅通过mongosh连接到集群。

db.getSiblingDB("cluster_changes").events.find()

[
  {
    _id: {
      _data: '826A8F5E65000000012B042C0100296E5A10049CC3BB749CB641F4BF5A032A2ABE2FB1463C6F7065726174696F6E54797065003C696E736572740046646F63756D656E744B65790046645F6964006466C1A1F1F1F1F1F1F1F1F1F1000004'
    },
    clusterTime: Timestamp({ t: 1787780709, i: 1 }),
    documentKey: { _id: ObjectId('66c1a1f1f1f1f1f1f1f1f1f1') },
    fullDocument: { _id: ObjectId('66c1a1f1f1f1f1f1f1f1f1f1'), title: 'Example Movie' },
    ns: { db: 'sample_mflix', coll: 'movies' },
    operationType: 'insert',
    wallTime: ISODate('2026-08-26T21:45:09.059Z')
  }
]

输出文档中的 ns字段显示该更改源自源集群上的 sample_mflix.movies 插入。变更流聚合管道会将这一更改反映到目标集群上的 cluster_changes.events集合中。

文档数组示例

以下聚合使用内联文档数组作为流媒体数据源，其中包含三个位置的天气观测值。该数组使用与示例天气数据集相同的模式。此聚合运行三个阶段：

$source 阶段将天气观测的内联 documents数组定义为流媒体数据源，并使用 timeField 将每个文档的 timestamp字段指定为权威时间戳。
$match 阶段仅将 dewPoint.value 大于 5.0 的文档传递到下一阶段。
$merge 阶段将输出写入 sample_weatherstream 数据库中名为 stream 的 Atlas 集合。如果不存在此类数据库或集合，Atlas 会创建它们。

[{
  "$source": {
    "documents": [
      {
        "location": "New York",
        "timestamp": ISODate('2024-01-15T08:00:00Z'),
        "temp": 23.5,
        "dewPoint": { "value": 6.2 }
      },
      {
        "location": "Los Angeles",
        "timestamp": ISODate('2024-01-15T08:05:00Z'),
        "temp": 18.2,
        "dewPoint": { "value": 4.8 }
      },
      {
        "location": "Chicago",
        "timestamp": ISODate('2024-01-15T08:10:00Z'),
        "temp": 26.8,
        "dewPoint": { "value": 7.5 }
      }
    ]
  }
},
{
  "$match": { "dewPoint.value": { "$gt": 5.0 } }
},
{
  "$merge": {
    "into": {
      "connectionName": "weatherStreamOutput",
      "db": "sample_weatherstream",
      "coll": "stream"
    }
  }
}]

要查看生成的 sample_weatherstream.stream 集合中的文档，请连接到您的 Atlas 集群并运行以下命令：

db.getSiblingDB("sample_weatherstream").stream.find()

[
  {
    _id: ObjectId('67a6c3df14fcac13b1a21a01'),
    dewPoint: { value: 6.2 },
    location: 'New York',
    temp: 23.5,
    timestamp: ISODate('2024-01-15T08:00:00.000Z')
  },
  {
    _id: ObjectId('67a6c3df14fcac13b1a21a03'),
    dewPoint: { value: 7.5 },
    location: 'Chicago',
    temp: 26.8,
    timestamp: ISODate('2024-01-15T08:10:00.000Z')
  }
]

Solar Stream 示例

以下聚合筛选来自示例流媒体数据源sample_stream_solar 的报告。它将结果存档到 solar-cluster-connection 连接。要学习；了解如何创建流处理工作区并添加与Atlas Stream Processing工作区的连接，请参阅开始使用Atlas Stream Processing和添加Atlas Stream Processing连接。此聚合运行三个阶段来过滤来自 sample_stream_solar 源的报告，并将结果写入solarDb数据库中名为 solarColl 的集合：

$source 阶段连接到 sample_stream_solar 源。 timeField字段使用 $dateFromString 将每个传入报告的 timestamp字段转换为日期。
$match 阶段会排除 device_id 为 device_8 的文档，并将来自所有其他设备的报告传递到下一阶段。
$merge 阶段将输出写入 solar-cluster-connection集群的 solarDb数据库中的 solarColl集合。

[{
  "$source": {
    "connectionName": "sample_stream_solar",
    "timeField": {
      "$dateFromString": { "dateString": "$timestamp" }
    }
  }
},
{
  "$match": { "device_id": { "$ne": "device_8" } }
},
{
  "$merge": {
    "into": {
      "connectionName": "solar-cluster-connection",
      "db": "solarDb",
      "coll": "solarColl"
    }
  }
}]

注意

sample_stream_solar 源是一种仅供测试的连接，每秒都会生成示例文档，用于快速原型设计。

针对 solar-cluster-connection集群运行以下命令，验证流处理器的行为。要进行连接，请参阅通过mongosh连接到集群。

要查看生成的 solarDb.solarColl集合中的文档，运行：

db.getSiblingDB("solarDb").solarColl.find()

[
  {
    _id: ObjectId('6a8f5f0352c6ffe64c8884ad'),
    device_id: 'device_3',
    event_type: 0,
    group_id: 8,
    max_watts: 450,
    obs: { watts: 428, temp: 10 },
    timestamp: '2024-08-12T21:47:47.287+00:00'
  }
]

每个文档都包含太阳能设备报告的 device_id、event_type、group_id、max_watts、obs 和 timestamp 字段。您可以观察到输出中没有文档的 device_id 为 device_8，因为 $match 阶段在 $merge 阶段将剩余文档写入 solarColl 之前排除了这些报告。

注意

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

Cron 计划示例

以下聚合每五分钟生成一个文档，而不是从连接中读取。此聚合运行三个阶段：

$source 阶段每五分钟的第二个 0 发出一个空文档（UTC 时间）。
$project 阶段使用作业名称标记每个文档，并将计划时间戳从 _ts字段复制到 runAt字段。
$merge 阶段将输出写入 sample_weatherstream 数据库中名为 heartbeats 的 Atlas 集合。如果不存在此类数据库或集合，Atlas 会创建它们。

[{
  "$source": {
    "schedule": "0 0/5 * * * *"
  }
},
{
  "$project": {
    "job": "five-minute-heartbeat",
    "runAt": "$_ts"
  }
},
{
  "$merge": {
    "into": {
      "connectionName": "weatherStreamOutput",
      "db": "sample_weatherstream",
      "coll": "heartbeats"
    }
  }
}]

要查看生成的 sample_weatherstream.heartbeats 集合中的文档，请连接到您的 Atlas 集群并运行以下命令：

db.getSiblingDB("sample_weatherstream").heartbeats.find()

[
  {
    _id: ObjectId('68b8a1c014fcac13b1a21b01'),
    job: 'five-minute-heartbeat',
    runAt: ISODate('2026-09-03T14:00:00.000Z')
  },
  {
    _id: ObjectId('68b8a2f414fcac13b1a21b02'),
    job: 'five-minute-heartbeat',
    runAt: ISODate('2026-09-03T14:05:00.000Z')
  },
  {
    _id: ObjectId('68b8a42814fcac13b1a21b03'),
    job: 'five-minute-heartbeat',
    runAt: ISODate('2026-09-03T14:10:00.000Z')
  }
]

注意

上述是一个具有代表性的示例。您看到的时间戳取决于您启动流处理器的时间。

$source 阶段（流处理）

定义

语法

Apache Kafka 代理

MongoDB 集合变更流

MongoDB database change stream

MongoDB 集群范围变更流源

AWS Kinesis Data Stream

文档数组

Cron 计划

行为

示例

Kafka 示例

变更流示例

数据库变更流示例

集群范围的变更流示例

文档数组示例

Solar Stream 示例

Cron 计划示例

`$source` 阶段（流处理）