$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 venir después de tu $source etapa y antes de tu 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 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"
      }
    }
  }
}