/ /

`$merge` （流处理）

定义

The $merge stage specifies a connection in the Connection Registry to write messages to. The connection must be an Atlas connection.

展示位置

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

语法

The Atlas Stream Processing version of $merge uses most of the same fields as the Atlas Data Federation version. Atlas Stream Processing also uses the following fields that are either unique to its implementation of $merge, or modified to suit it. To learn more about the fields shared with Atlas Data Federation $merge, see $merge syntax.

$merge 管道阶段采用以下原型形式：

{
  "$merge": {
    "into": {
      "connectionName": "<registered-atlas-connection>",
      "db": "<registered-database-name>" | <expression>,
      "coll": "<atlas-collection-name>" | <expression>
    },
    "on": "<identifier field>" | [ "<identifier field1>", ...],
    "let": {
      <var_1>: <expression>,
      <var_2>: <expression>,
      …,
      <var_n>: <expression>
    },
    "whenMatched": "replace | keepExisting | merge |  delete | <pipeline> | <expression>",
    "whenNotMatched": "insert | discard | expression",
    "parallelism": <integer>
  }
}

字段	必要性	说明
`into`	必需	简化以反映仅支持 `$merge` 到 Atlas 连接的 Atlas 流处理。要了解更多信息，请参阅对 Atlas Data Federation `$merge` 字段的说明。
`whenMatched`	Optional	与 Atlas Data Federation `$merge` 阶段相比，扩展了功能，支持 `"delete"` 和动态表达式。当设置为 `"delete"` 时，Atlas 会从目标集合中删除与条件匹配的所有消息。如果您使用动态表达式值，它必须解析为以下字符串之一： `"merge"` `"replace"` `"keepExisting"` `"delete"` `"<pipeline>"` `"<expression>"`
`whenNotMatched`	Optional	与 Atlas Data Federation `$merge` 阶段相比，扩展了功能，支持动态表达式。如果您使用动态表达式值，它必须解析为以下字符串之一： `"insert"` `"discard"` `"expression"`
`parallelism`	可选的	用于分配写入操作的线程数量。默认为 `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。

行为

限制

The on field has special requirements for $merge against sharded collections. To learn more, see $merge syntax.

$merge can't write to time series collections. To write documents to time series collections, use the $emit stage.

您需要具有Atlas管理员角色才能在分片的集合上使用 $merge。

动态表达式

You can use a field or dynamic expression as the value of the following fields:

into.db
into.coll

这使您的流处理器能够逐条消息将消息写入不同的目标 Atlas collection。

例子

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

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

要将其中每一个分类到不同的 Atlas 数据库和collection中，您可以编写以下$merge阶段：

$merge: {
  into: {
    connectionName: "db1",
    db: "$customerStatus",
    coll: "$transactionType"
  }
}

此$merge阶段：

将Very Important Industries消息写入名为VIP.subscription的 Atlas collection。
将N. E. Buddy消息写入名为employee.requisition的 Atlas collection。
将Khan Traktor消息写入名为contractor.billableHours的 Atlas collection。

You can only use dynamic expressions that evaluate to strings. For more information on dynamic expressions, see expression operators.

If you specify a database or collection 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 主题的流媒体数据保存到 Atlas 集群中的集合中，请使用 $merge 阶段和$source阶段。$source 阶段指定了从中读取数据的主题。$merge 阶段会将此数据写入目标集合。

使用以下语法：

{
  "$source": {
    "connectionName": "<registered-kafka-connection>",
    "topic": [ "<topic-name-1>", "<topic-name-2>", ... ]
  }
},
{
  "$merge": {
    "into": {
      "connectionName": "<registered-atlas-connection>",
      "db": "<registered-database-name>" | <expression>,
      "coll": "<atlas-collection-name>" | <expression>
      }
    },
    ...
}

示例

基本示例

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 a dewPoint.value of less than or equal to 5.0 and passes along the documents with dewPoint.value greater than 5.0 to the next stage.
The $merge stage writes the output to an Atlas collection named stream in the sample_weatherstream database. If no such database or collection exist, Atlas creates them.

{
  '$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'),
  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'
  }
}

注意

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

跨Atlas集群复制变更流事件

You can use the $merge.whenMatched and $merge.whenNotMatched parameters to replicate the effects of Change Stream events according to their operation type.

以下聚合有四个阶段：

$source 阶段通过 atlas1 连接在 Atlas 集群上建立与 db1.coll1 集合的连接。
The $addFields stage enriches the ingested documents with a fullDocument._isDelete field set to the value of an equality check between the "$operationType value of each document and "delete". This equality evaluates to a boolean.
$replaceRoot 阶段将文档替换为增强的 $fullDocument 字段的值。
$merge 阶段通过 atlas2 连接写入 db1.coll1，对每个文档执行两项检查：
- 首先，whenMatched 字段检查文档是否与 db1.coll1 集合中通过 _id 匹配的现有文档匹配，这是默认的匹配字段，因为 on 未显式设置。如果条件成立，并且 fullDocument._isDelete 设置为 true，那么 Atlas 会删除匹配的文档。如果匹配并且 fullDocument._isDelete 设置为 false，则 Atlas 会用流数据源中的新文档替换匹配的文档。
- 其次，如果 Atlas Stream Processing 没有找到匹配的文档并且 fullDocument._isDelete 为 true，Atlas 会丢弃该文档，而不是将其写入集合。如果没有此类匹配文档并且 fullDocument._isDelete 为 false，Atlas 会将流数据源中的文档插入到集合中。

{
  $source: {
    connectionName: “atlas1”,
    db: “db1”,
    coll: “coll1”,
    fullDocument: “required”
  }
},
{
  $addFields: {
    “fullDocument._isDelete”: {
      $eq: [
        “$operationType”,
        “delete”
      ]
    }
  }
},
{
  $replaceRoot: {
    newRoot: “$fullDocument”
  }
},
{
  $merge: {
    into: {
      connectionName: “atlas2”,
      db: “db1”,
      coll: “coll1”
    },
    whenMatched: {
      $cond: {
        if: “$_isDelete”,
        then: “delete”,
        else: “replace”
      }
    },
    whenNotMatched: {
      $cond: {
        if: “$_isDelete”,
        then: “discard”,
        else: “insert”
      }
    },
  }
}

$merge （流处理）

定义