/ /

`$lookup` （流处理）

定义

The $lookup stage performs a left outer join of the stream of messages from your $source to an Atlas collection in your Connection Registry.

根据您的使用案例，$lookup 管道阶段将使用以下三种语法之一：

要了解更多信息，请参阅$lookup 语法。

警告

使用 $lookup 扩充流可能会降低流处理速度。

以下原型表单展示了所有可用字段：

{
  "$lookup": {
    "from": {
      "connectionName": "<registered-atlas-connection>",
      "db": "<registered-database-name>",
      "coll": "<atlas-collection-name>",
      "readPreference": "<read-preference>",
      "readPreferenceTags": [
        {"<key>": "<value>"},
        ...
      ]
    },
    "localField": "<field-in-source-messages>",
    "foreignField": "<field-in-from-collection>",
    "let": {
      <var_1>: <expression>,
      <var_2>: <expression>,
      …,
      <var_n>: <expression>
    },
    "pipeline": [
      <pipeline to run>
    ],
    "as": "<output-array-field>",
    "parallelism": <integer>,
    "partitionBy": <expression>
  }
}

语法

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

字段	类型	必要性	说明
from	文档	可选的	用于指定Atlas数据库中要加入到来自`$source` 的消息的集合的文档。您必须仅指定连接注册表中的集合。如果指定此字段，则必须指定此文档中所有字段的值。如果您指定 `pipeline`字段，则不需要此字段。
from.connectionName	字符串	可选的	连接注册表中的连接名称。如果您指定 `pipeline`字段，则不需要此字段。
from.db	字符串	可选的	包含您要加入的集合的 Atlas 数据库名称。如果您指定 `pipeline`字段，则不需要此字段。
from.coll	字符串	可选的	您想要加入的集合的名称。如果您指定 `pipeline`字段，则不需要此字段。
from.readPreference	字符串	Optional	Read preference for `$lookup` operations. 默认值为 `primary`。
from.readPreferenceTags	阵列	Optional	Read preference tags for `$lookup` operations.
localField	字符串	可选的	要联接的 `$source` 信息中的字段。该字段是以下语法的一部分：涉及单个条件联接的等值匹配采用简洁语法的关联子查询
foreignField	字符串	可选的	要联接的 `from` 集合中文档的字段。该字段是以下语法的一部分：涉及单个条件联接的等值匹配采用简洁语法的关联子查询
let	文档	可选的	指定各个管道阶段中使用的变量。要了解详情，请参阅 let。该字段是以下语法的一部分：已联接集合上的联接条件和子查询采用简洁语法的关联子查询
管道	文档	可选的	Specifies the `pipeline` to run on the joined collection. To learn more, see pipeline. 该字段是以下语法的一部分：已联接集合上的联接条件和子查询采用简洁语法的关联子查询
作为	字符串	必需	要添加到输入文档中的新数组字段的名称。这个新数组字段包含 `from` 集合中的匹配文档。如果指定名称已作为字段存在于输入文档中，则该字段将被覆盖。
`parallelism`	整型	Optional	向 `$lookup` 目标发出的最大并行请求数。较高的并行度值可能会增加 `$lookup`吞吐量。但是，较高的值也可能会在 `$lookup` 目标集群上使用更多的资源。必须是介于 `1` 和 `64` 之间的整数。默认为 `1`。每个流处理器都有一个由其层级确定的最大累积并行度值。流处理器的累积并行度计算如下： `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。
`partitionBy`	表达式(expression)	Optional	用于将输入流划分为并行线程的表达式。Atlas Stream Processing将具有相同 `partitionBy`表达式结果的每个输入文档分配给同一线程。如果未指定此字段，则输入文档将通过循环发送到并行线程。

行为

Atlas Stream Processing 版本的 $lookup 对来自 $source 的消息和指定 Atlas 集合中的文档执行左外连接。此版本的行为类似于标准 MongoDB 数据库中提供的 $lookup 阶段。但是，此版本要求您将连接注册表中的 Atlas 集合指定为 from 字段的值。

The pipeline can contain a nested $lookup stage. If you include a nested $lookup stage in your pipeline, you must use the standard from syntax to specify a collection in the same remote Atlas connection as the outer $lookup stage.

例子

$lookup : {
  from: {connectionName: "dbsrv1", db: "db1", coll: "coll1"},
    …,
  pipeline: [
    …,
    {
      $lookup: {
        from: "coll2",
        …,
      }
    },
    …,
  ]
}

If your pipeline has both $lookup and $merge on the same collection, Atlas Stream Processing results might vary if you try to maintain an incremental view. Atlas Stream Processing processes multiple source messages simultaneously and then merges them all together. If multiple messages have the same ID, which both $lookup and $merge use, Atlas Stream Processing might return results that haven't yet materialized.

例子

考虑以下输入流：

{ _id: 1, count: 2 }
{ _id: 1, count: 3 }

假设查询包含管道内的以下内容：

{
  ...,
  pipeline: [
    { $lookup on _id == foreignDoc._id from collection A }
    { $project: { _id: 1, count: $count + $foreignDoc.count } }
    { $merge: { into collection A } }
  ]
}

如果尝试保持增量视图，则可能会看到类似于以下内容的结果：

{ _id: 1, count: 5 }

但是，Atlas Stream Processing 可能会返回 5 或 3 的计数，具体取决于 Atlas Stream Processing 是否处理了文档。

有关更多信息，请参阅 $lookup。

示例

A streaming data source generates detailed weather reports from various locations, conformant to the schema of the Sample Weather Dataset. A collection named humidity_descriptions contains documents of the form:

其中，relative_humidity 字段描述了室温（ 20 摄氏度）下的相对湿度， condition列出了适合该湿度水平的语言描述符。您可以使用 $lookup 阶段，用推荐的描述符来丰富流媒体天气报告，方便气象学家在天气预报广播中使用。

以下聚合有四个阶段：

$source 阶段与 Apache Kafka 代理建立连接，在名为 my_weatherdata 的主题中收集这些报告，从而在将每条记录引入后续聚合阶段时将其公开。此阶段还会覆盖其投影的时间戳字段的名称，将其设置为 ingestionTime。
$lookup 阶段将 humidity_descriptions 数据库中的记录合并到 dewPoint 字段的天气报告中。
The $match stage excludes documents that have an empty humidity_info field, and passes along documents with a populated humidity_info field to the next stage.
The $merge stage writes the output to an Atlas collection named enriched_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'
  }
},
{
  '$lookup': {
    from: {
      connectionName: 'weatherStream',
      db: 'humidity',
      coll: 'humidity_descriptions'
    },
  'localField':'dewPoint.value',
  'foreignField':'dewPoint',
  'as': 'humidity_info'
  }
}
{ '$match': { 'humidity_info': { '$ne': [] } } }
{
  '$merge': {
    into: {
      connectionName: 'weatherStream',
      db: 'sample_weatherstream',
      coll: 'enriched_stream'
    }
  }
}

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

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

{
  st: 'x+55100+006100',
  position: {
    type: 'Point',
    coordinates: [
      92.7,
      -53.6
    ]
  },
  elevation: 9999,
  callLetters: 'UECN',
  qualityControlProcess: 'V020',
  dataSource: '4',
  type: 'FM-13',
  airTemperature: {
    value: -11,
    quality: '9'
  },
  dewPoint: {
    value: 12.5,
    quality: '1'
  },
  pressure: {
    value: 1032.7,
    quality: '9'
  },
  wind: {
    direction: {
      angle: 300,
      quality: '9'
    },
    type: '9',
    speed: {
      rate: 23.6,
      quality: '2'
    }
  },
  visibility: {
    distance: {
      value: 14000,
      quality: '1'
    },
    variability: {
      value: 'N',
      quality: '1'
    }
  },
  skyCondition: {
    ceilingHeight: {
      value: 390,
      quality: '9',
      determination: 'C'
    },
    cavok: 'N'
  },
  sections: [
    'SA1',
    'AA1',
    'OA1',
    'AY1',
    'AG1'
  ],
  precipitationEstimatedObservation: {
    discrepancy: '4',
    estimatedWaterDepth: 21
  },
  atmosphericPressureChange: {
    tendency: {
      code: '1',
      quality: '1'
    },
    quantity3Hours: {
      value: 5.5,
      quality: '1'
    },
    quantity24Hours: {
      value: 99.9,
      quality: '9'
    }
  },
  seaSurfaceTemperature: {
    value: 1.3,
    quality: '9'
  },
  waveMeasurement: {
    method: 'M',
    waves: {
      period: 4,
      height: 2.5,
      quality: '9'
    },
    seaState: {
      code: '00',
      quality: '9'
    }
  },
  pastWeatherObservationManual: {
    atmosphericCondition: {
      value: '4',
      quality: '1'
    },
    period: {
      value: 6,
      quality: '1'
    }
  },
  skyConditionObservation: {
    totalCoverage: {
      value: '07',
      opaque: '99',
      quality: '1'
    },
    lowestCloudCoverage: {
      value: '06',
      quality: '1'
    },
    lowCloudGenus: {
      value: '07',
      quality: '9'
    },
    lowestCloudBaseHeight: {
      value: 2250,
      quality: '9'
    },
    midCloudGenus: {
      value: '07',
      quality: '9'
    },
    highCloudGenus: {
      value: '00',
      quality: '1'
    }
  },
  presentWeatherObservationManual: {
    condition: '75',
    quality: '1'
  },
  atmosphericPressureObservation: {
    altimeterSetting: {
      value: 9999.9,
      quality: '9'
    },
    stationPressure: {
      value: 1032.6,
      quality: '1'
    }
  },
  skyCoverLayer: {
    coverage: {
      value: '09',
      quality: '1'
    },
    baseHeight: {
      value: 240,
      quality: '9'
    },
    cloudType: {
      value: '99',
      quality: '9'
    }
  },
  liquidPrecipitation: {
    period: 6,
    depth: 3670,
    condition: '9',
    quality: '9'
  },
  extremeAirTemperature: {
    period: 99.9,
    code: 'N',
    value: -30.9,
    quantity: '9'
  },
  ingestionTime: ISODate('2024-09-19T20:04:34.346Z'),
  humidity_info: [
    {
      _id: ObjectId('66ec805ad3cfbba767ebf7a5'),
      dewPoint: 12.5,
      relativeHumidity: 62,
      condition: 'humid, muggy'
    }
  ],
}

注意

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

$lookup （流处理）

定义

语法

行为

示例

`$lookup` （流处理）