$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 | 为传入消息定义权威时间戳的文档。 如果使用
如果您未声明 | |
tsFieldName | 字符串 | Optional | 覆盖 $source 投影的时间戳字段名称的名称。 Atlas Stream Processing 管道中的 $source 阶段使用分配的文档时间戳投影一个名为 | |
partitionIdleTimeout | 文档 | Optional | 指定分区在水印计算中被忽略之前允许其空闲的时间长度的文档。 | |
partitionIdleTimeout.size | 整型 | Optional | 指定分区空闲超时持续时间的数字。 | |
partitionIdleTimeout.unit | 字符串 | Optional | 分区空闲超时持续时间的时间单位。
| |
config | 文档 | Optional | 包含可覆盖各种默认值的字段的文档。 | |
config.auto_offset_reset | 字符串 | Optional | 指定 Apache Kafka 中的哪个事件 开始摄取的源主题。
默认值为 | |
config.group_id | 字符串 | Optional | 与流处理器关联的 kafka 使用者群组的 ID。 如果省略,Atlas Stream Processing 会将流处理实例与以下格式的自动生成的 ID 相关联:
Atlas Stream Processing 将分区偏移提交到 Apache Kafka 提交检查点后指定消费者组 ID 的代理。当直到偏移量之前的消息持久记录在检查点中时,它才会提交偏移量。 这样,您就可以直接从 Kafka 代理消费者组元数据跟踪流处理器的偏移延迟和进度。 | |
config.keyFormat | 字符串 | Optional | 用于反序列化 Apache Kafka 的数据类型 关键数据。必须是以下值之一:
默认值为 | |
config.keyFormatError | 字符串 | Optional | 如何处理反序列化 Apache Kafka 时遇到的错误 关键数据。必须是以下值之一:
|
注意
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 | 为传入消息定义权威时间戳的文档。 如果使用
如果您未声明 |
tsFieldName | 字符串 | Optional | 覆盖源声明的默认时间戳字段名称的名称。 Atlas Stream Processing 管道在内部向传入消息添加一个名为 |
db | 字符串 | 必需 | 在 connectionName 指定的 Atlas 实例上托管的 MongoDB database 的名称。此数据库的change stream充当流媒体数据源。 |
coll | 字符串或字符串数组 | 必需 | 在 connectionName 指定的 Atlas 实例上托管的一个或多个 MongoDB 集合的名称。这些集合的变更流充当流数据源。如果省略此字段,流处理器将源自MongoDB 数据库变更流。 |
config | 文档 | Optional | 包含可覆盖各种默认值的字段的文档。 |
config.startAfter | 令牌 | 可选的 | 源开始报告后发生的变更事件。它采用恢复令牌的形式。 您只能使用 |
config.startAtOperationTime | timestamp | 可选的 | 源应开始报告的操作时间。 您只能使用 |
config.fullDocument | 字符串 | 可选的 | 用于控制变更流源是应返回完整文档还是仅在发生更新时返回变更的设置。 必须是以下之一:
如果没有为 fullDocument 指定值,则默认为 要将此字段与集合变更流一起使用,您必须在该集合上启用变更流前图像和后图像。 |
config.fullDocumentOnly | 布尔 | 可选的 | 用于控制change stream源是返回包括所有元数据的整个事件文档,还是仅返回 要将此字段与集合变更流一起使用,您必须在该集合上启用变更流前图像和后图像。 |
config.fullDocumentBeforeChange | 字符串 | Optional | 指定change stream源是否应在输出中包含处于原始“更改之前”状态的文档。必须是以下之一:
如果没有为 要将此字段与集合变更流一起使用,您必须在该集合上启用变更流前图像和后图像。 |
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 | 为传入消息定义权威时间戳的文档。 如果使用
如果您未声明 |
tsFieldName | 字符串 | Optional | 覆盖源声明的默认时间戳字段名称的名称。 Atlas Stream Processing 管道在内部向传入消息添加一个名为 |
db | 字符串 | 必需 | 在 connectionName 指定的 Atlas 实例上托管的 MongoDB database 的名称。此数据库的change stream充当流媒体数据源。 |
config | 文档 | Optional | 包含可覆盖各种默认值的字段的文档。 |
config.startAfter | 令牌 | 可选的 | 源开始报告后发生的变更事件。它采用恢复令牌的形式。 您只能使用 |
config.startAtOperationTime | timestamp | 可选的 | 源应开始报告的操作时间。 您只能使用 |
config.fullDocument | 字符串 | 可选的 | 用于控制变更流源是应返回完整文档还是仅在发生更新时返回变更的设置。 必须是以下之一:
如果没有为 fullDocument 指定值,则默认为 要将此字段与数据库变更流一起使用,您必须对该数据库中的每个集合启用变更流前像和后像。 |
config.fullDocumentOnly | 布尔 | 可选的 | 用于控制change stream源是返回包括所有元数据的整个事件文档,还是仅返回 要将此字段与数据库变更流一起使用,您必须对该数据库中的每个集合启用变更流前像和后像。 |
config.fullDocumentBeforeChange | 字符串 | Optional | 指定change stream源是否应在输出中包含处于原始“更改之前”状态的文档。必须是以下之一:
如果没有为 要将此字段与数据库变更流一起使用,您必须对该数据库中的每个集合启用变更流前像和后像。 |
config.pipeline | 文档 | Optional | 指定聚合管道以在源点筛选变更流输出。此管道必须符合change-stream-modify-output 中描述的参数。 |
文档数组
为了对文档数组进行操作, $source
阶段具有以下原型形式:
{ "$source": { "timeField": { $toDate | $dateFromString: <expression> }, "tsFieldName": "<timestamp>", "documents" : [{source-doc},...] | <expression> } }
$source
阶段采用包含以下字段的文档:
字段 | 类型 | 必要性 | 说明 |
---|---|---|---|
timeField | 文档 | Optional | 为传入消息定义权威时间戳的文档。 如果使用
如果您未声明 |
tsFieldName | 字符串 | Optional | 覆盖源声明的默认时间戳字段名称的名称。 Atlas Stream Processing 管道在内部向传入消息添加一个名为 |
documents | 阵列 | 可选的 | 用作流媒体数据源的文档数组。该字段的值可以是对象数组,也可以是计算结果为对象数组的表达式。使用 connectionName 字段时请勿使用此字段。 |