`$source`

在此页面上

定义

语法
Apache Kafka 代理
MongoDB 集合变更流
MongoDB database change stream
MongoDB集群范围的变更流源
文档数组
行为
示例

定义

$source

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

Apache Kafka 代理
MongoDB collection change stream
MongoDB database change stream
文档数组

注意

您不能将 Atlas 无服务器实例用作$source 。

语法

Apache Kafka 代理

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

{
  "$source": {
    "connectionName": "<registered-connection>",
    "topic" : "<source-topic>",
    "timeField": {
      $toDate | $dateFromString: <expression>
    },
    "tsFieldName": "<timestamp>",
    "partitionIdleTimeout": {
      "size": <duration-number>,
      "unit": "<duration-unit>"
    },
    "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 会根据源提供的消息时间戳创建一个时间戳。

tsFieldName

字符串

Optional

覆盖 $source 投影的时间戳字段名称的名称。

Atlas Stream Processing 管道中的 $source 阶段使用分配的文档时间戳投影一个名为_ts的字段。流数据源还可能使用名为_ts的字段来存储每条消息的时间戳。为防止这些字段之间发生冲突，请在进行其他处理之前使用tsFieldName重命名源提供的任何名为_ts的字段。

partitionIdleTimeout

文档

Optional

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

partitionIdleTimeout.size

整型

Optional

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

partitionIdleTimeout.unit

字符串

Optional

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

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

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

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 会将流处理实例与以下格式的自动生成的 ID 相关联：

asp-${streamProcessorId}-consumer

Atlas Stream Processing 将分区偏移提交到 Apache Kafka 提交检查点后指定消费者组 ID 的代理。当直到偏移量之前的消息持久记录在检查点中时，它才会提交偏移量。这样，您就可以直接从 Kafka 代理消费者组元数据跟踪流处理器的偏移延迟和进度。

config.keyFormat

字符串

Optional

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

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

默认值为 binData。

config.keyFormatError

字符串

Optional

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

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

注意

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

MongoDB 集合变更流

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

{
  "$source": {
    "connectionName": "<registered-connection>",
    "timeField": {
      $toDate | $dateFromString: <expression>
    },
    "tsFieldName": "<timestamp>",
    "db" : "<source-db>",
    "coll" : ["<source-coll>",...],
    "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 会根据源提供的消息时间戳创建一个时间戳。
`tsFieldName`	字符串	Optional	覆盖源声明的默认时间戳字段名称的名称。 Atlas Stream Processing管道在内部向传入消息添加一个名为`_ts`的字段，用于存储时间戳信息。流媒体数据源还可能使用名为`_ts`的字段来存储每条消息的时间戳。为防止这些字段之间发生冲突，请在进行其他处理之前使用`tsFieldName`重命名源提供的任何名为`_ts`的字段。
`db`	字符串	必需	在`connectionName`指定的 Atlas 实例上托管的 MongoDB database 的名称。此数据库的change stream充当流媒体数据源。
`coll`	字符串或字符串数组	必需	在`connectionName`指定的 Atlas 实例上托管的一个或多个 MongoDB 集合的名称。这些集合的变更流充当流数据源。如果省略此字段，流处理器将源自MongoDB 数据库变更流。
`config`	文档	Optional	包含可覆盖各种默认值的字段的文档。
`config.startAfter`	token	可选的	源开始报告后发生的变更事件。它采用恢复令牌的形式。您只能使用`config.startAfter`或`config.StartAtOperationTime`之一。
`config.startAtOperationTime`	timestamp	可选的	源应开始报告的操作时间。您只能使用`config.startAfter`或`config.StartAtOperationTime`之一。
`config.fullDocument`	字符串	可选的	用于控制变更流源是应返回完整文档还是仅在发生更新时返回变更的设置。必须是以下之一： `updateLookup` ：仅返回更新时的更改。 `required` ：必须返回完整的文档。如果没有完整文档，则不返回任何内容。 `whenAvailable` ：只要有完整文档可用，就返回完整文档，否则返回更改。如果没有为 fullDocument 指定值，则默认为`updateLookup` 。要将此字段与集合变更流一起使用，您必须在该集合上启用变更流前图像和后图像。
`config.fullDocumentOnly`	布尔	可选的	用于控制change stream源是返回包括所有元数据的整个事件文档，还是仅返回`fullDocument`内容的设置。如果设置为`true` ，源仅返回`fullDocument`的内容。要将此字段与集合变更流一起使用，您必须在该集合上启用变更流前图像和后图像。
`config.fullDocumentBeforeChange`	字符串	Optional	指定change stream源是否应在输出中包含处于原始“更改之前”状态的文档。必须是以下之一： `off` ：省略`fullDocumentBeforeChange`字段。 `required` ：必须以更改前的状态返回完整文档。如果处于更改前状态的完整文档不可用，则流处理器将失败。 `whenAvailable` ：只要有可用的文档，就返回处于更改前状态的完整文档，否则忽略`fullDocumentBeforeChange`字段。如果没有为`fullDocumentBeforeChange`指定值，则默认为`off` 。要将此字段与集合变更流一起使用，您必须在该集合上启用变更流前图像和后图像。
`config.pipeline`	文档	Optional	指定聚合管道以在源点筛选变更流输出。此管道必须符合change-stream-modify-output 中描述的参数。

MongoDB database change stream

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

{
  "$source": {
    "connectionName": "<registered-connection>",
    "timeField": {
      $toDate | $dateFromString: <expression>
    },
    "tsFieldName": "<timestamp>",
    "db" : "<source-db>",
    "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 会根据源提供的消息时间戳创建一个时间戳。
`tsFieldName`	字符串	Optional	覆盖源声明的默认时间戳字段名称的名称。 Atlas Stream Processing管道在内部向传入消息添加一个名为`_ts`的字段，用于存储时间戳信息。流媒体数据源还可能使用名为`_ts`的字段来存储每条消息的时间戳。为防止这些字段之间发生冲突，请在进行其他处理之前使用`tsFieldName`重命名源提供的任何名为`_ts`的字段。
`db`	字符串	必需	在`connectionName`指定的 Atlas 实例上托管的 MongoDB database 的名称。此数据库的change stream充当流媒体数据源。
`config`	文档	Optional	包含可覆盖各种默认值的字段的文档。
`config.startAfter`	token	可选的	源开始报告后发生的变更事件。它采用恢复令牌的形式。您只能使用`config.startAfter`或`config.StartAtOperationTime`之一。
`config.startAtOperationTime`	timestamp	可选的	源应开始报告的操作时间。您只能使用`config.startAfter`或`config.StartAtOperationTime`之一。
`config.fullDocument`	字符串	可选的	用于控制变更流源是应返回完整文档还是仅在发生更新时返回变更的设置。必须是以下之一： `updateLookup` ：仅返回更新时的更改。 `required` ：必须返回完整的文档。如果没有完整文档，则不返回任何内容。 `whenAvailable` ：只要有完整文档可用，就返回完整文档，否则返回更改。如果没有为 fullDocument 指定值，则默认为`updateLookup` 。要将此字段与数据库变更流一起使用，您必须对该数据库中的每个集合启用变更流前像和后像。
`config.fullDocumentOnly`	布尔	可选的	用于控制change stream源是返回包括所有元数据的整个事件文档，还是仅返回`fullDocument`内容的设置。如果设置为`true` ，源仅返回`fullDocument`的内容。要将此字段与数据库变更流一起使用，您必须对该数据库中的每个集合启用变更流前像和后像。
`config.fullDocumentBeforeChange`	字符串	Optional	指定change stream源是否应在输出中包含处于原始“更改之前”状态的文档。必须是以下之一： `off` ：省略`fullDocumentBeforeChange`字段。 `required` ：必须以更改前的状态返回完整文档。如果处于更改前状态的完整文档不可用，则流处理器将失败。 `whenAvailable` ：只要有可用的文档，就返回处于更改前状态的完整文档，否则忽略`fullDocumentBeforeChange`字段。如果没有为`fullDocumentBeforeChange`指定值，则默认为`off` 。要将此字段与数据库变更流一起使用，您必须对该数据库中的每个集合启用变更流前像和后像。
`config.pipeline`	文档	Optional	指定聚合管道以在源点筛选变更流输出。此管道必须符合change-stream-modify-output 中描述的参数。

MongoDB集群范围的变更流源

为了操作来自整个Atlas 集群变更流的流媒体数据， $source阶段具有以下原型形式：

{
  "$source": {
    "connectionName": "<registered-connection>",
    "timeField": {
      $toDate | $dateFromString: <expression>
    },
    "tsFieldName": "<timestamp>",
    "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 会根据源提供的消息时间戳创建一个时间戳。
`tsFieldName`	字符串	Optional	覆盖源声明的默认时间戳字段名称的名称。 Atlas Stream Processing管道在内部向传入消息添加一个名为`_ts`的字段，用于存储时间戳信息。流媒体数据源还可能使用名为`_ts`的字段来存储每条消息的时间戳。为防止这些字段之间发生冲突，请在进行其他处理之前使用`tsFieldName`重命名源提供的任何名为`_ts`的字段。
`config`	文档	Optional	包含可覆盖各种默认值的字段的文档。
`config.startAfter`	token	可选的	源开始报告后发生的变更事件。它采用恢复令牌的形式。您只能使用`config.startAfter`或`config.StartAtOperationTime`之一。
`config.startAtOperationTime`	timestamp	可选的	源应开始报告的操作时间。您只能使用`config.startAfter`或`config.StartAtOperationTime`之一。
`config.fullDocument`	字符串	可选的	用于控制变更流源是应返回完整文档还是仅在发生更新时返回变更的设置。必须是以下之一： `updateLookup` ：仅返回更新时的更改。 `required` ：必须返回完整的文档。如果没有完整文档，则不返回任何内容。 `whenAvailable` ：只要有完整文档可用，就返回完整文档，否则返回更改。如果没有为 fullDocument 指定值，则默认为`updateLookup` 。要将此字段与数据库变更流一起使用，您必须对该数据库中的每个集合启用变更流前像和后像。
`config.fullDocumentOnly`	布尔	可选的	用于控制change stream源是返回包括所有元数据的整个事件文档，还是仅返回`fullDocument`内容的设置。如果设置为`true` ，源仅返回`fullDocument`的内容。要将此字段与数据库变更流一起使用，您必须对该数据库中的每个集合启用变更流前像和后像。
`config.fullDocumentBeforeChange`	字符串	Optional	指定change stream源是否应在输出中包含处于原始“更改之前”状态的文档。必须是以下之一： `off` ：省略`fullDocumentBeforeChange`字段。 `required` ：必须以更改前的状态返回完整文档。如果处于更改前状态的完整文档不可用，则流处理器将失败。 `whenAvailable` ：只要有可用的文档，就返回处于更改前状态的完整文档，否则忽略`fullDocumentBeforeChange`字段。如果没有为`fullDocumentBeforeChange`指定值，则默认为`off` 。要将此字段与数据库变更流一起使用，您必须对该数据库中的每个集合启用变更流前像和后像。
`config.pipeline`	文档	Optional	指定聚合管道以在源点筛选变更流输出。此管道必须符合change-stream-modify-output 中描述的参数。

文档数组

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

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

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

字段	类型	必要性	说明
`timeField`	文档	Optional	为传入消息定义权威时间戳的文档。如果使用`timeField` ，则必须将其定义为以下之一：将源消息字段作为参数的`$toDate`表达式将源消息字段作为参数的`$dateFromString`表达式。如果您未声明`timeField` ，Atlas Stream Processing 会根据源提供的消息时间戳创建一个时间戳。
`tsFieldName`	字符串	Optional	覆盖源声明的默认时间戳字段名称的名称。 Atlas Stream Processing管道在内部向传入消息添加一个名为`_ts`的字段，用于存储时间戳信息。流媒体数据源还可能使用名为`_ts`的字段来存储每条消息的时间戳。为防止这些字段之间发生冲突，请在进行其他处理之前使用`tsFieldName`重命名源提供的任何名为`_ts`的字段。
`documents`	阵列	可选的	用作流媒体数据源的文档数组。该字段的值可以是对象数组，也可以是计算结果为对象数组的表达式。使用`connectionName`字段时请勿使用此字段。

行为

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

示例

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

$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',
    tsFieldName: 'ingestionTime'
  }
},
{ '$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'),
  _stream_meta: {
    source: {
      type: 'kafka',
      topic: 'my_weatherdata',
      partition: 0,
      offset: Long('165235')
    }
  },
  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'
  }
}

注意

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

后退

聚合管道

来年

$validate