$setStreamMeta Stage de agregación (Stream Processing)

Definición

$setStreamMeta

La fase $setStreamMeta establece campos de metadatos para los documentos entrantes. Puedes utilizar estos metadatos para realizar operaciones selectivas en documentos basados en valores específicos sin alterar el contenido de los propios documentos.

Sintaxis

Un $setStreamMeta etapa tiene la siguiente forma prototipo:

{
  $setStreamMeta: {
    "<metadata-field>": <expression>
  }
}

La etapa $setStreamMeta procesa un documento con los siguientes campos:

Comportamiento

$setStreamMeta debe ir después de la etapa $source y antes de la etapa $emit o $merge.

Ejemplos

Una fuente de datos de transmisión genera un documento por cada pedido realizado en una plataforma de comercio electrónico. Los documentos adoptan la siguiente forma:

{
  orderId: 1,
  productId: "A",
  qty: 2
}

Una base de datos de Atlas contiene una colección products con documentos de la siguiente forma:

[
  {_id: "A", name: "Laptop", category: "tech"},
  {_id: "B", name: "Shirt", category: "clothing"}
]

La siguiente agregación tiene seis etapas:

1. La etapa $source recopila documentos de pedidos de la plataforma de comercio electrónico, exponiendo cada registro a medida que se recopila a las siguientes etapas de agregación.

2. La etapa $lookup une los documentos del pedido entrante a una products colección en la shop base de datos, en el _id campo. El campo de matriz resultante se product denomina.

3. La etapa $unwind convierte el valor del arreglo de un solo elemento de product en un documento.

4. La etapa establece un campo de metadatos $setStreamMeta llamado stream.coll igual al valor product.category de.

5. La etapa $unset remueve el campo product, devolviendo el documento de origen a su forma original.

6. La etapa $merge fusiona el documento en una colección determinada dinámicamente por el valor de stream.coll.

{
  $source: {
    documents: [
      {
        orderId: 1,
        productId: "A",
        qty: 2
      },
      {
        orderId: 2,
        productId: "B",
        qty: 1
      },
      {
        orderId: 3,
        productId: "A",
        qty: 5
      }
    ]
  }
},
{
  $lookup: {
    from: {
      connectionName: "atlas",
      db: "shop",
      coll: "products"
    },
    localField: "productId",
    foreignField: "_id",
    as: "product"
  }
},
{
  $unwind: "$product"
},
{
  $setStreamMeta: {
    "stream.coll": "$product.category"
  }
},
{
  $unset: [
    "product"
  ]
},
{
  $merge: {
    into: {
      connectionName: "atlas",
      db: "shop",
      coll: {
        $meta: "stream.coll"
      }
    }
  }
}