对于 AI 代理:可在 https://www.mongodb.com/zh-cn/docs/llms.txt 获取文档索引—通过在任何 URL 路径后添加 .md 可获取所有页面的 Markdown 版本。

$cachedLookup

$cachedLookup

$cachedLookup 阶段执行从 $source 到连接注册表中的Atlas集合的消息流的左外连接。

此阶段的功能与 $lookup 阶段类似,但根据可配置参数缓存查询结果。

重要

$cachedLookup 不支持let 或 pipeline 字段。

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

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

{ "$cachedLookup": { "ttl": { "size": <int>, "unit": "ms" | "second" | "minute" | "hour" | "day" }, "maxMemUsageBytes": <int>, "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>", "as": "<output-array-field>" }}

$cachedLookup 使用一些与 $lookup 的通用版本相同的字段。$cachedLookup 包括用于配置查询缓存行为的字段,并为 from 字段提供修改后的语法,用于通过来自连接注册表的连接查询数据。

字段
类型
必要性
说明

TTL

文档

必需

指定缓存查询的TTL的文档。

ttl.size

int

必需

units 中缓存的查询的TTL大小。

ttl.unit

字符串

必需

测量缓存查询的TTL的时间单位。必须是以下之一:

  • "ms"

  • "second"

  • "minute"

  • "hour"

  • "day"

maxMemUsageBytes

int

必需

分配给查询缓存的最大内存(以字节为单位)。如果缓存的大小超过此值, Atlas Stream Processing先将较旧的结果逐出以释放空间。如果过期结果的数量不足以低于此阈值,Atlas Stream Processing会随机逐出缓存的查询,直到缓存大小低于阈值。

默认值为流处理工作区中可用 RAM 的 10%。您在流处理工作区中将 maxMemUsageBytes 设定为不超过可用 RAM 的 12.5%。

from

文档

必需

用于指定Atlas数据库中要加入到来自$source 的消息的集合的文档。您必须仅指定连接注册表中的集合。

如果指定此字段,则必须指定此文档中所有字段的值。

from.connectionName

字符串

必需

连接注册表中的连接名称。

from.db

字符串

必需

包含您要加入的集合的 Atlas 数据库名称。

from.coll

字符串

必需

您想要加入的集合的名称。

from.readPreference

字符串

Optional

对 from 集合上的查询的读取偏好(read preference)。

默认值为 primary。

from.readPreferenceTags

阵列

Optional

读取偏好标签,用于对 from 集合的查询。

localField

字符串

必需

要联接的 $source 信息中的字段。

foreignField

字符串

必需

要联接的 from 集合中文档的字段。

作为

字符串

必需

要添加到输入文档中的新数组字段的名称。这个新数组字段包含 from 集合中的匹配文档。如果指定的名称已作为字段存在于输入文档中,则现有字段将被覆盖。

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

此外,$cachedLookup 会在可配置的时间长度内缓存查询结果。使用此功能查询不常更改的数据,以提高效率。当缓存条目的TTL到期时, Atlas Stream Processing会逐出该条目。如果在您进行新查询时缓存条目的总大小等于 maxMemoryUsageBytes,Atlas Stream Processing会逐出条目,直到有空间来缓存新查询。

流数据源从不同位置生成详细的天气报告,符合 示例天气数据集的模式。名为 humidity_descriptions 的集合包含以下形式的文档:

{ 'dew_point': 16.2, 'relative_humidity': 79, 'condition': 'sticky, oppressive'}

其中,relative_humidity字段描述了室温(20 摄氏度)下的相对湿度,condition 列出了适合该湿度水平的语言描述符。您可以使用$cachedLookup阶段通过建议的描述符来丰富流媒体天气报告,以供气象学家在天气广播中使用。

以下聚合有四个阶段:

  1. $source 阶段与 Apache Kafka 代理建立连接,在名为 my_weatherdata 的主题中收集这些报告,从而在将每条记录引入后续聚合阶段时将其公开。此阶段还会覆盖其投影的时间戳字段的名称,将其设置为 ingestionTime。

  2. $cachedLookup 阶段将 humidity_descriptions数据库中的记录连接到 dewPoint字段的天气报告中。每个查询的TTL为 5 minute, Atlas Stream Processing最多可存储 200 MB 的结果。

  3. $match 阶段会排除 humidity_info 字段为空的文档,并将 humidity_info 字段已填充的文档传递到下一阶段。

  4. $merge 阶段将输出写入 sample_weatherstream 数据库中名为 enriched_stream 的 Atlas 集合。如果不存在此类数据库或集合,Atlas 会创建它们。

{ '$source': { connectionName: 'sample_weatherdata', topic: 'my_weatherdata', tsFieldName: 'ingestionTime' }},{ '$cachedLookup': { "ttl": { "size": 5, "unit": "minute" }, "maxMemUsageBytes": 209715200, 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' } ],}

注意

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

给本页内容打分