/ /

Características

Pipeline de agregación

Pipeline de agregación

`$merge` Etapa (Atlas Data Federation)

$merge Escribe los resultados de una canalización de agregación en una colección temporal en el clúster de Atlas. A continuación, Atlas Data Federation ejecuta $merge localmente en el clúster de 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:

Guarde datos desde cualquiera de los almacenes de instancia federada de base de datos compatibles.
Guardar en el mismo o en otro clúster de Atlas, base de datos o colección dentro del mismo Proyecto de Atlas.

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

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 clúster de 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 de Atlas. Este es el ID del Proyecto que contiene su instancia federada de base de datos. Si se omite, se usa como valor por defecto el ID del Proyecto que contiene la instancia federada de base de datos.	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 tu fichero o tus almacenes de datos en tu clúster de Atlas, tus documentos podrían tener campos duplicados _id. Esta sección describe cómo Atlas Data Federation resuelve los duplicados e incluye recomendaciones para resolver los duplicados en tu pipeline de agregación.

Resolviendo IDS Duplicados en Atlas Data Federation

Para resolver los 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, elije cuidadosamente el valor para las opciones whenMatched y whenNotMatched de la etapa $merge.

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

Supón que ejecutas el siguiente pipeline en los documentos listados arriba:

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.

Supón que ejecutas el siguiente pipeline en los documentos listados arriba:

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.

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

Supón que ejecutas el siguiente pipeline en los documentos listados arriba:

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

Atlas Data Federation luego fusiona este documento en la colección objetivo en el clúster Atlas especificado.

Ejemplo

Combinar 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 )

Ejecutar `$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	})

$merge Etapa (Atlas Data Federation)

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

Combinar datos usando $merge

Ejemplo

Ejecutar $merge en segundo plano

Ejemplo

`$merge` Etapa (Atlas Data Federation)

Combinar datos usando `$merge`

Ejecutar `$merge` en segundo plano