/ /

`$lookup` (Processamento de Stream)

Definição

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

Dependendo do seu caso de uso, um estágio de pipeline $lookup usa uma das três sintaxes a seguir:

Para saber mais, consulte Sintaxe $lookup.

Aviso

Usar $lookup para enriquecimento de um fluxo pode reduzir a velocidade de processamento do fluxo.

O seguinte formulário protótipo ilustra todos os campos disponíveis:

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

Sintaxe

O estágio $lookup recebe um documento com os seguintes campos:

Campo	Tipo	necessidade	Descrição
from	documento	Condicional	Documento que especifica uma collection em um banco de dados Atlas para unir às mensagens do seu `$source`. Você deve especificar uma collection somente a partir do seu Registro de Conexão. Se você especificar este campo, você deverá especificar valores para todos os campos neste documento. Este campo não é exigido se você especificar um campo `pipeline`.
from.connectionName	string	Condicional	Nome da conexão no registro de conexões. Este campo não é exigido se você especificar um campo `pipeline`.
from.db	string	Condicional	Nome do banco de dados do Atlas que contém a coleção que você deseja unir. Este campo não é exigido se você especificar um campo `pipeline`.
from.coll	string	Condicional	Nome da coleção da qual você deseja participar. Este campo não é exigido se você especificar um campo `pipeline`.
from.readPreference	string	Opcional	Read preference for `$lookup` operations. Padrão é `primary`.
from.readPreferenceTags	array	Opcional	Read preference tags for `$lookup` operations.
localField	string	Condicional	Campo a partir de suas mensagens `$source` do qual participar. Este campo faz parte das seguintes sintaxes: Correspondência de igualdade com uma coleção de junção única. Subconsultas correlacionadas usando sintaxe concisa
foreignField	string	Condicional	Campo de documentos na coleção `from` para participar. Este campo faz parte das seguintes sintaxes: Correspondência de igualdade com uma coleção de junção única. Subconsultas correlacionadas usando sintaxe concisa
let	documento	Condicional	Especifica as variáveis a serem usadas nos estágios do pipeline. Para saber mais, consulte let. Este campo faz parte das seguintes sintaxes: Condições de junção e subconsultas em uma coleção associada Subconsultas correlacionadas usando sintaxe concisa
gasoduto	documento	Condicional	Specifies the `pipeline` to run on the joined collection. To learn more, see pipeline. Este campo faz parte das seguintes sintaxes: Condições de junção e subconsultas em uma coleção associada Subconsultas correlacionadas usando sintaxe concisa
como	string	Obrigatório	Nome do novo campo de array a ser adicionado aos documentos de entrada. Este novo campo de array contém os documentos correspondentes da coleção `from`. Se o nome especificado já existir como campo no documento de entrada, esse campo será substituído.
`parallelism`	inteiro	Opcional	Número máximo de solicitações paralelas feitas para o destino `$lookup`. Valores de paralelismo mais altos podem aumentar a taxa de transferência `$lookup`. No entanto, valores mais altos também podem usar mais recursos no cluster de destino `$lookup` . Deve ser um número inteiro entre `1` e `64`. O padrão é `1`. Cada processador de fluxo tem um valor máximo de paralelismo cumulativo determinado por seu nível. O paralelismo cumulativo de um processador de fluxo é calculado da seguinte forma: `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`. Por exemplo, se o seu estágio `$source` define um valor `parallelism` de `4`, o seu estágio `$lookup` não define nenhum valor `parallelism` (portanto, o padrão é `1`) e o seu estágio `$merge` define um `parallelism` valor de `2`, então você tem dois `parallelized stages`, e o paralelismo cumulativo do seu processador de fluxo é calculado como `(4 + 2) - 2`. Se um processador de fluxo exceder o paralelismo cumulativo máximo para seu nível, o Atlas Stream Processing lançará um erro e o avisará sobre o nível mínimo de processador necessário para o nível pretendido de paralelismo. Você deve dimensionar o processador para um nível superior ou reduzir os valores de paralelismo de seus estágios para resolver o erro. Para saber mais, consulte Stream Processing.
`partitionBy`	expressão	Opcional	Expressão usada para dividir o fluxo de entrada em threads paralelos. O Atlas Stream Processing aloca cada documento de entrada com o mesmo resultado de expressão `partitionBy` para o mesmo thread. Se este campo não for especificado, os documentos de entrada serão enviados para threads paralelos por round-robin.

Comportamento

A versão do Atlas Stream Processing de $lookup executa um left join externo de mensagens do seu $source e os documentos em uma coleção especificada do Atlas. Essa versão se comporta de forma semelhante ao estágio $lookup disponível em um banco de dados MongoDB padrão. No entanto, esta versão exige que você especifique uma coleção Atlas do seu Registro de Conexão como o valor para o campo 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.

Exemplo

$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.

Exemplo

Considere o seguinte fluxo de entrada:

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

Suponha que sua query contenha o seguinte dentro do pipeline:

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

Se estiver tentando manter uma visualização gradual, o resultado poderá ser semelhante ao seguinte:

{ _id: 1, count: 5 }

No entanto, o Atlas Stream Processing pode retornar uma contagem de 5 ou 3, dependendo se o Atlas Stream Processing processou os documentos.

Para mais informações, consulte $lookup.

Exemplos

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:

Onde o campo relative_humidity descreve a umidade relativa em temperatura ambiente (20 graus Celsius), e condition lista descritores verbais adequados para esse nível de umidade. Você pode usar a fase $lookup para enriquecer os relatórios meteorológicos em tempo real com descritores sugeridos para meteorologistas usarem nas transmissões meteorológicas.

A seguinte agregação tem quatro fases:

O estágio $source estabelece uma conexão com o broker do Apache Kafka coletando esses relatórios em um tópico chamado my_weatherdata, expondo cada registro à medida que ele é assimilado nos estágios de agregação subsequentes. Esse estágio também substitui o nome do campo de carimbo de data/hora que ele projeta, definindo-o como ingestionTime.
O estágio $lookup junta os registros do banco de dados humidity_descriptions aos relatórios meteorológicos no campo 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'
    }
  }
}

Para visualizar os documentos na coleção sample_weatherstream.enriched_stream resultante, conecte-se ao cluster Atlas e execute o seguinte comando:

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'
    }
  ],
}

Observação

O exemplo anterior é representativo. Os dados de streaming não são estáticos e cada usuário vê documentos distintos.

Voltar

$HTTPS

$cachedLookup

$lookup (Processamento de Stream)

Definição

Sintaxe

Comportamento

Exemplos

`$lookup` (Processamento de Stream)