$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 de los pedidos entrantes a una colección de products en la base de datos shop en el campo _id. El campo de arreglo resultante se llama product.

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

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

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