/ /

Pipeline de agregación

Docs Home

/ /

Características

Pipeline de agregación

Pipeline de agregación

`$merge` Etapa (Federación de Datos Atlas)

$merge escribe 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:

Escribe datos desde cualquiera de los almacenes de instancias de bases de datos federadas compatibles.
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 la escritura en un clúster Atlas, Atlas Data Federation introduce una sintaxis alternativa para el into campo obligatorio. Atlas Data Federation admite todos los demás campos, como se describe $merge para.

Para obtener más información, consulte la etapa de $merge canalización.

Permisos requeridos

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

encontrar y
insert

Considerations

Si la agregación falla, Atlas Data Federation no revierte ninguna escritura que completó antes de que ocurriera el $merge 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 escribir 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 otroson letcampos,,, whenMatched y,whenNotMatched consulte la documentación del servidor MongoDB $merge para.

Nota

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

opciones

Opción

Tipo

Descripción

Necesidad

background

booleano

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

{ "background" : true }

Utilice esta opción si desea enviar otras consultas nuevas sin esperar a que se completen las consultas que se están ejecutando actualmente o desconectar su conexión de instancia de base de datos federada mientras las consultas continúan ejecutándose en segundo plano.

Opcional

Resolución de identificaciones 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.

Resolución de identificaciones duplicadas 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 duplicado _id y todos los documentos subsiguientes en una nueva colección Atlas Y.
Ejecuta la etapa $merge especificada para fusionar la colección Y en la colección X.
Escribe los documentos resultantes en la colección de destino en el clúster Atlas especificado.

Nota

Atlas Data Federation solo resuelve 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 _id duplicados, puede:

Incluya una etapa en su canalización para especificar el orden en que Atlas Data Federation debe procesar los documentos $sort 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 los duplicados durante la etapa $merge cuando la opción whenMatched está establecida 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 denominadas X y Y:

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

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

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

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

Luego, 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 denominadas 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 con la colección X. Para la opción whenMatched: replace en la canalización, Atlas Data Federation reemplaza el documento _id: 1 en la colección X por el documento _id: 1 en la colección Y. Por lo tanto, el resultado de la canalización con duplicados contiene los siguientes documentos:

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

Luego, 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á establecida en discard utilizando 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 denominadas 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 la canalización, Atlas Data Federation reemplaza el documento _id: 1 en la colección X por el documento _id: 1 en la colección Y. Para la opción whenNotMatched: discard en la canalización, Atlas Data Federation descarta los documentos de la colección Y que no coinciden con un documento de la colección X. Por lo tanto, el resultado de la canalización con duplicados contiene solo 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`

La siguiente sintaxis del ejemplo $merge escribe los resultados en una sampleDB.mySampleData colección del clúster Atlas myTestCluster llamado. El ejemplo no especifica un ID de proyecto; la $merge etapa utiliza el ID del proyecto que contiene la 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	})

Permisos requeridos

Considerations

Sintaxis

Campos

Nota

opciones

Resolución de identificaciones de documentos duplicados

Resolución de identificaciones duplicadas en Atlas Data Federation

Nota

Remediación de IDs duplicados

Ejemplo

Ejemplo

Ejemplo

Fusionar datos usando $merge

Ejemplo

Ejecutar $merge en segundo plano

Ejemplo

Fusionar datos usando `$merge`

Ejecutar `$merge` en segundo plano