/ /

Pipeline de agregación

Docs Home

/ /

Características

Pipeline de agregación

Pipeline de agregación

`$merge` Etapa (Atlas Data Federation)

$merge guarda los resultados de un Canal de agregación a una colección temporal en el clúster Atlas. Atlas Data Federation ejecuta $merge localmente en el clúster Atlas para fusionar los datos en fragmentos en la colección de destino. En caso de fallo durante una $merge operación, esto garantiza que al menos una parte de los datos se escriba en la colección de destino.

En Atlas Data Federation, $merge puede:

Guardar datos de cualquiera de los compatibles de instancias de bases de datos federadas.
Escribir en el mismo clúster, base de datos o colección de Atlas o en uno diferente dentro del mismo proyecto de Atlas.

Para permitir escrituras en un clúster de Atlas, Atlas Data Federation introduce sintaxis alternativa para el campo into requerido. Atlas Data Federation admite todos los demás campos según se describe para $merge.

Para obtener más información, consulte $merge la etapa del pipeline.

Permisos requeridos

Para usar $merge para escribir en una colección del clúster de Atlas, debes ser un usuario de base de datos con los siguientes privilegios:

encontrar y
insert

Considerations

Si la agregación falla, Atlas Data Federation no revierte ninguna escritura que la $merge completó antes de que ocurriera el error.

Sintaxis

{
  "$merge": {
    "into": {
      "atlas": {
        "projectId": "<atlas-project-ID>",
        "clusterName": "<atlas-cluster-name>",
        "db": "<atlas-database-name>",
        "coll": "<atlas-collection-name>"
      }
    },
    "on": "<identifier field>"|[ "<identifier field1>", ...],
    "let": { <variables> },
    "whenMatched": "replace|keepExisting|merge|fail|pipeline",
    "whenNotMatched": "insert|discard|fail"
  }
}

Campos

Esta sección describe la sintaxis alternativa que Atlas Data Federation proporciona para el campo into.

Campo	Tipo	Descripción	Necesidad
`atlas`	Objeto	Ubicación para guardar los documentos desde el pipeline de agregación.	Requerido
`clusterName`	string	Nombre del cluster Atlas.	Requerido
`coll`	string	Nombre de la colección en el clúster Atlas.	Requerido
`db`	string	Nombre de la base de datos en el clúster Atlas que contiene la colección.	Requerido
`projectId`	string	Identificador único del proyecto que contiene el clúster Atlas. Este es el ID del proyecto que contiene la instancia de base de datos federada. Si se omite, se usa el ID predeterminado del proyecto que contiene la instancia de base de datos federada.	Opcional

Para obtener más información sobre los otros campos, on, let, whenMatched, y whenNotMatched, consulta la documentación del servidor de MongoDB para $merge.

Nota

Para utilizar on en varios campos, debes crear un índice único compuesto en los campos identificadores ``on``.

opciones

Opción

Tipo

Descripción

Necesidad

background

booleano

Bandera para ejecutar operaciones de agregación en segundo plano. Si se omite, el valor por defecto es false. Cuando se establece en true, Atlas Data Federation ejecuta las consultas en segundo plano.

{ "background" : true }

Utiliza esta opción si deseas enviar otras consultas nuevas sin esperar a que se completen las consultas en curso o desconectar la conexión de tu instancia federada de base de datos mientras las consultas continúan ejecutándose en segundo plano.

Opcional

Resolución de IDs de documentos duplicados

Al escribir documentos desde su archivo o almacenes de datos en su clúster de Atlas, es posible que tengan campos _id duplicados. Esta sección describe cómo Atlas Data Federation resuelve los duplicados e incluye recomendaciones para resolverlos en su canal de agregación.

Resolviendo IDS Duplicados en Atlas Data Federation

Para resolver duplicados, Atlas Data Federation:

Escribe documentos en una colección de Atlas X en el orden en que recibe los documentos hasta que encuentra un duplicado.
Escribe el documento con el campo _id duplicado y todos los documentos posteriores en una nueva colección de Atlas Y.
Ejecuta la etapa especificada $merge para fusionar la colección Y en la colección X.
Guarda los documentos resultantes en la colección de destino en el clúster de Atlas especificado.

Nota

Atlas Data Federation solo resuelve los valores duplicados en el campo _id. No resuelve valores duplicados en otros campos con índices únicos.

Remediación de IDs duplicados

Para remediar los campos duplicados de _id, se puede:

Include una $sort etapa en su pipeline para especificar el orden en que Atlas Data Federation debe procesar los documentos resultantes.

Según el orden de los documentos que fluyen hacia la etapa,$merge elija whenMatched whenNotMatched cuidadosamente el valor para $merge las opciones y de la etapa.

Ejemplo

Los siguientes ejemplos muestran cómo Atlas Data Federation resuelve duplicados durante la etapa $merge cuando whenMatched la opción está configurada en keepExisting o replace. Estos ejemplos utilizan los siguientes documentos:

{
  "_id" : 1,
  "state" : "FL"
},
{
  "_id" : 1,
  "state" : "NJ"
},
{
  "_id" : 2,
  "state" : "TX"
}

Supongamos que ejecuta la siguiente canalización en los documentos enumerados anteriormente:

db.s3coll.aggregate([
  {
    "$sort": {
      "_id": 1,
      "state": 1,
    }
  },
  {
    "$merge": {
      "into": {
        "atlas": {
          "clusterName": "clustername",
          "db": "clusterdb",
          "coll": "clustercoll"
        }
      },
      "on": "_id",
      "whenMatched": "keepExisting",
      "whenNotMatched": "insert"
    }
  }
])

Atlas Data Federation escribe los siguientes datos en dos colecciones llamadas X y Y:

{
  "_id" : 1,
  "state" : "FL"
}

{
  "_id" : 1,
  "state" : "NJ"
},
{
  "_id" : 2,
  "state" : "TX"
}

Atlas Data Federation luego fusiona documentos de la colección Y en la colección X. Para la opción de whenMatched: keepExisting en el pipeline, Atlas Data Federation mantiene el documento existente con _id: 1 en la colección X. Por lo tanto, el resultado de la pipeline con duplicados contiene los siguientes documentos:

{
  "_id" : 1,
  "state" : "FL"
},
{
  "_id" : 2,
  "state" : "TX"
}

Atlas Data Federation fusiona estos documentos en la colección de destino en el clúster Atlas especificado.

Supongamos que ejecuta la siguiente canalización en los documentos enumerados anteriormente:

db.s3coll.aggregate([
  {
    "$sort": {
      "_id": 1,
      "state": 1,
    }
  },
  {
    "$merge": {
      "into": {
        "atlas": {
          "clusterName": "clustername",
          "db": "clusterdb",
          "coll": "clustercoll"
        }
      },
      "on": "_id",
      "whenMatched": "replace",
      "whenNotMatched": "insert"
    }
  }
])

Atlas Data Federation escribe los siguientes datos en dos colecciones llamadas X y Y:

{
  "_id" : 1,
  "state" : "FL"
}

{
  "_id" : 1,
  "state" : "NJ"
},
{
  "_id" : 2,
  "state" : "TX"
}

Atlas Data Federation fusiona documentos de la colección Y en la colección X. Para la opción whenMatched: replace en la pipeline, Atlas Data Federation reemplaza el documento con _id: 1 en la colección X con el documento con _id: 1 en la colección Y. Por lo tanto, el resultado de la pipeline con duplicados contiene los siguientes documentos:

{
  "_id" : 1,
  "state" : "NJ"
},
{
  "_id" : 2,
  "state" : "TX"
}

Atlas Data Federation fusiona estos documentos en la colección de destino en el clúster Atlas especificado.

Evite utilizar la opción whenNotMatched: discard.

Ejemplo

Este ejemplo muestra cómo Atlas Data Federation resuelve duplicados cuando la opción whenNotMatched está configurada en discard usando los siguientes documentos:

{
  "_id" : 1,
  "state" : "AZ"
},
{
  "_id" : 1,
  "state" : "CA"
},
{
  "_id" : 2,
  "state" : "NJ"
},
{
  "_id" : 3,
  "state" : "NY"
},
{
  "_id" : 4,
  "state" : "TX"
}

Supongamos que ejecuta la siguiente canalización en los documentos enumerados anteriormente:

db.archivecoll.aggregate([
  {
    "$sort": {
      "_id": 1,
      "state": 1,
    }
  },
  {
    "$merge": {
      "into": {
        "atlas": {
          "clusterName": "clustername",
          "db": "clusterdb",
          "coll": "clustercoll"
        }
      },
      "on": "_id",
      "whenMatched": "replace",
      "whenNotMatched": "discard"
    }
  }
])

Atlas Data Federation escribe los siguientes datos en dos colecciones llamadas X y Y:

{
  "_id" : 1,
  "state" : "AZ" // gets replaced
}

{
  "_id" : 1,
  "state" : "CA"
}
{
  "_id" : 2,
  "state" : "NJ" // gets discarded
}
{
  "_id" : 3,
  "state" : "NY" // gets discarded
}
{
  "_id" : 4,
  "state" : "TX" // gets discarded
}

Atlas Data Federation fusiona documentos de la colección Y en la colección X. Para la opción whenMatched: replace en el pipeline, Atlas Data Federation reemplaza el documento con _id: 1 en la colección X con el documento con _id: 1 en la colección Y. Para la opción whenNotMatched: discard en el pipeline, Atlas Data Federation descarta documentos en la colección Y que no coincidan con un documento en la colección X. Por lo tanto, el resultado de la pipeline con duplicados contiene únicamente el siguiente documento:

{
  "_id" : 1,
  "state" : "CA"
}

Luego, Atlas Data Federation fusiona este documento en la colección de destino en el clúster Atlas especificado.

Ejemplo

Fusionar datos usando `$merge`

El siguiente ejemplo de sintaxis $merge escribe los resultados en una colección sampleDB.mySampleData en el clúster Atlas denominada myTestCluster. El ejemplo no especifica un ID del grupo; la etapa $merge usa el ID del Proyecto que contiene tu instancia de base de datos federada.

Ejemplo

1 db.mySampleData.aggregate(
2   [
3     {
4       "$merge": {
5         "into": {
6           "atlas": {
7             "clusterName": "myTestCluster",
8             "db": "sampleDB",
9             "coll": "mySampleData"
10           }
11         },
12         ...
13       }
14     }
15   ]
16 )

Ejecute `$merge` en segundo plano

El siguiente ejemplo de sintaxis $merge escribe los resultados en una colección sampleDB.mySampleData en el clúster Atlas llamado myTestCluster en segundo plano. En el ejemplo no se especifica un ID del grupo; la etapa $merge utiliza el ID del Proyecto que contiene tu instancia federada de base de datos.

Ejemplo

1 db.runCommand({
2   "aggregate": "mySampleData",
3   "pipeline": [
4     {
5       "$merge": {
6         "into": {
7           "atlas": {
8             "clusterName": "myTestCluster",
9             "db": "sampleDB",
10             "coll": "mySampleData"
11           }
12         },
13         ...
14       }
15     }
16   ],
17   { "background" : true }
18 })

Volver

$lookup

$out

1	db.mySampleData.aggregate(
2	[
3	{
4	"$merge": {
5	"into": {
6	"atlas": {
7	"clusterName": "myTestCluster",
8	"db": "sampleDB",
9	"coll": "mySampleData"
10	}
11	},
12	...
13	}
14	}
15	]
16	)

1	db.runCommand({
2	"aggregate": "mySampleData",
3	"pipeline": [
4	{
5	"$merge": {
6	"into": {
7	"atlas": {
8	"clusterName": "myTestCluster",
9	"db": "sampleDB",
10	"coll": "mySampleData"
11	}
12	},
13	...
14	}
15	}
16	],
17	{ "background" : true }
18	})

Permisos requeridos

Considerations

Sintaxis

Campos

Nota

opciones

Resolución de IDs de documentos duplicados

Resolviendo IDS Duplicados en Atlas Data Federation

Nota

Remediación de IDs duplicados

Ejemplo

Ejemplo

Ejemplo

Fusionar datos usando $merge

Ejemplo

Ejecute $merge en segundo plano

Ejemplo

Fusionar datos usando `$merge`

Ejecute `$merge` en segundo plano