/ /

Realice consultas de snapshots de larga duración

Las consultas instantáneas le permiten leer datos tal como aparecieron en un solo punto en el tiempo en el pasado reciente.

A partir de MongoDB 5.0, puede usar la preocupación de lectura "snapshot"para consultar datos en nodos secundarios. Esta función aumenta la versatilidad y la resiliencia de las lecturas de su aplicación. No necesita crear una copia estática de sus datos, trasladarla a un sistema independiente ni aislar manualmente estas consultas de larga duración para que no interfieran con su carga de trabajo operativa. En su lugar, puede realizar consultas de larga duración en una base de datos transaccional activa mientras lee datos consistentes.

El uso de la preocupación de "snapshot" lectura en nodos secundarios no afecta la carga de trabajo de escritura de la aplicación. Solo las lecturas de la aplicación se benefician del aislamiento de las consultas de larga duración en los nodos secundarios.

Utilice consultas de instantáneas cuando desee:

Realice múltiples consultas relacionadas y asegúrese de que cada consulta lea datos del mismo momento.
Asegúrese de leer un estado consistente de los datos de algún momento del pasado.

Comparación de las preocupaciones sobre lectura local e instantánea

Cuando MongoDB realiza consultas de larga duración utilizando la consulta de lectura predeterminada, los resultados de la consulta pueden contener datos de escrituras simultáneas. Por lo tanto, la consulta puede devolver resultados inesperados o "local" incoherentes.

Para evitar esta situación, cree una sesión y especifique la preocupación de lectura. Con "snapshot" "snapshot" la preocupación de lectura, MongoDB ejecuta la consulta con aislamiento de instantáneas, lo que significa que la consulta lee los datos tal como aparecieron en un momento específico del pasado reciente.

Ejemplos

Los ejemplos de esta página muestran cómo puedes usar consultas instantáneas para:

Ejecutar consultas relacionadas desde el mismo momento
Leer desde un estado consistente de los datos de algún punto en el pasado

Ejecutar consultas relacionadas desde el mismo momento

La lectura de "snapshot" la inquietud le permite ejecutar múltiples consultas relacionadas dentro de una sesión y garantizar que cada consulta lea datos del mismo punto en el tiempo.

Un refugio de animales tiene una base de datos pets que contiene colecciones para cada tipo de mascota. La base de datos pets contiene estas colecciones:

cats
dogs

Cada documento de cada colección contiene un campo adoptable que indica si la mascota está disponible para adopción. Por ejemplo, un documento de la colección cats se ve así:

{
   "name": "Whiskers",
   "color": "white",
   "age": 10,
   "adoptable": true
}

Quiere ejecutar una consulta para ver el número total de mascotas disponibles para adopción en todas las colecciones. Para obtener una visión coherente de los datos, debe asegurarse de que los datos devueltos por cada colección correspondan a un único momento.

Para lograr este objetivo, utilice la preocupación "snapshot" de lectura dentro de una sesión:

mongoc_client_session_t *cs = NULL;
mongoc_collection_t *cats_collection = NULL;
mongoc_collection_t *dogs_collection = NULL;
int64_t adoptable_pets_count = 0;
bson_error_t error;
mongoc_session_opt_t *session_opts;
cats_collection = mongoc_client_get_collection(client, "pets", "cats");
dogs_collection = mongoc_client_get_collection(client, "pets", "dogs");
/* Seed 'pets.cats' and 'pets.dogs' with example data */
if (!pet_setup(cats_collection, dogs_collection)) {
   goto cleanup;
}
/* start a snapshot session */
session_opts = mongoc_session_opts_new();
mongoc_session_opts_set_snapshot(session_opts, true);
cs = mongoc_client_start_session(client, session_opts, &error);
mongoc_session_opts_destroy(session_opts);
if (!cs) {
   MONGOC_ERROR("Could not start session: %s", error.message);
   goto cleanup;
}
/*
 * Perform the following aggregation pipeline, and accumulate the count in
 * `adoptable_pets_count`.
 *
 *  adoptablePetsCount = db.cats.aggregate(
 *      [ { "$match": { "adoptable": true } },
 *        { "$count": "adoptableCatsCount" } ], session=s
 *  ).next()["adoptableCatsCount"]
 *
 *  adoptablePetsCount += db.dogs.aggregate(
 *      [ { "$match": { "adoptable": True} },
 *        { "$count": "adoptableDogsCount" } ], session=s
 *  ).next()["adoptableDogsCount"]
 *
 * Remember in order to apply the client session to
 * this operation, you must append the client session to the options passed
 * to `mongoc_collection_aggregate`, i.e.,
 *
 * mongoc_client_session_append (cs, &opts, &error);
 * cursor = mongoc_collection_aggregate (
 *    collection, MONGOC_QUERY_NONE, pipeline, &opts, NULL);
 */
accumulate_adoptable_count(cs, cats_collection, &adoptable_pets_count);
accumulate_adoptable_count(cs, dogs_collection, &adoptable_pets_count);
printf("there are %" PRId64 " adoptable pets\n", adoptable_pets_count);

using namespace mongocxx;
using bsoncxx::builder::basic::kvp;
using bsoncxx::builder::basic::make_document;
auto db = client["pets"];
int64_t adoptable_pets_count = 0;
auto opts = mongocxx::options::client_session{};
opts.snapshot(true);
auto session = client.start_session(opts);
{
    pipeline p;
    p.match(make_document(kvp("adoptable", true))).count("adoptableCatsCount");
    auto cursor = db["cats"].aggregate(session, p);
    for (auto doc : cursor) {
        adoptable_pets_count += doc.find("adoptableCatsCount")->get_int32();
    }
}
{
    pipeline p;
    p.match(make_document(kvp("adoptable", true))).count("adoptableDogsCount");
    auto cursor = db["dogs"].aggregate(session, p);
    for (auto doc : cursor) {
        adoptable_pets_count += doc.find("adoptableDogsCount")->get_int32();
    }
}

ctx := context.TODO()
sess, err := client.StartSession(options.Session().SetSnapshot(true))
if err != nil {
	return err
}
defer sess.EndSession(ctx)
var adoptablePetsCount int32
err = mongo.WithSession(ctx, sess, func(ctx context.Context) error {
	// Count the adoptable cats
	const adoptableCatsOutput = "adoptableCatsCount"
	cursor, err := db.Collection("cats").Aggregate(ctx, mongo.Pipeline{
		bson.D{{"$match", bson.D{{"adoptable", true}}}},
		bson.D{{"$count", adoptableCatsOutput}},
	})
	if err != nil {
		return err
	}
	if !cursor.Next(ctx) {
		return fmt.Errorf("expected aggregate to return a document, but got none")
	}
	resp := cursor.Current.Lookup(adoptableCatsOutput)
	adoptableCatsCount, ok := resp.Int32OK()
	if !ok {
		return fmt.Errorf("failed to find int32 field %q in document %v", adoptableCatsOutput, cursor.Current)
	}
	adoptablePetsCount += adoptableCatsCount
	// Count the adoptable dogs
	const adoptableDogsOutput = "adoptableDogsCount"
	cursor, err = db.Collection("dogs").Aggregate(ctx, mongo.Pipeline{
		bson.D{{"$match", bson.D{{"adoptable", true}}}},
		bson.D{{"$count", adoptableDogsOutput}},
	})
	if err != nil {
		return err
	}
	if !cursor.Next(ctx) {
		return fmt.Errorf("expected aggregate to return a document, but got none")
	}
	resp = cursor.Current.Lookup(adoptableDogsOutput)
	adoptableDogsCount, ok := resp.Int32OK()
	if !ok {
		return fmt.Errorf("failed to find int32 field %q in document %v", adoptableDogsOutput, cursor.Current)
	}
	adoptablePetsCount += adoptableDogsCount
	return nil
})
if err != nil {
	return err
}

db = client.pets
async with await client.start_session(snapshot=True) as s:
    adoptablePetsCount = 0
    docs = await db.cats.aggregate(
        [{"$match": {"adoptable": True}}, {"$count": "adoptableCatsCount"}], session=s
    ).to_list(None)
    adoptablePetsCount = docs[0]["adoptableCatsCount"]
    docs = await db.dogs.aggregate(
        [{"$match": {"adoptable": True}}, {"$count": "adoptableDogsCount"}], session=s
    ).to_list(None)
    adoptablePetsCount += docs[0]["adoptableDogsCount"]
print(adoptablePetsCount)

$catsCollection = $client->selectCollection('pets', 'cats');
$dogsCollection = $client->selectCollection('pets', 'dogs');
$session = $client->startSession(['snapshot' => true]);
$adoptablePetsCount = $catsCollection->aggregate(
    [
        ['$match' => ['adoptable' => true]],
        ['$count' => 'adoptableCatsCount'],
    ],
    ['session' => $session],
)->toArray()[0]->adoptableCatsCount;
$adoptablePetsCount += $dogsCollection->aggregate(
    [
        ['$match' => ['adoptable' => true]],
        ['$count' => 'adoptableDogsCount'],
    ],
    ['session' => $session],
)->toArray()[0]->adoptableDogsCount;
var_dump($adoptablePetsCount);

db = client.pets
with client.start_session(snapshot=True) as s:
    adoptablePetsCount = (
        (
            db.cats.aggregate(
                [{"$match": {"adoptable": True}}, {"$count": "adoptableCatsCount"}],
                session=s,
            )
        ).next()
    )["adoptableCatsCount"]
    adoptablePetsCount += (
        (
            db.dogs.aggregate(
                [{"$match": {"adoptable": True}}, {"$count": "adoptableDogsCount"}],
                session=s,
            )
        ).next()
    )["adoptableDogsCount"]
print(adoptablePetsCount)

client = Mongo::Client.new(uri_string, database: "pets")
client.start_session(snapshot: true) do |session|
  adoptable_pets_count = client['cats'].aggregate([
    { "$match": { "adoptable": true } },
    { "$count": "adoptable_cats_count" }
  ], session: session).first["adoptable_cats_count"]
  adoptable_pets_count += client['dogs'].aggregate([
    { "$match": { "adoptable": true } },
    { "$count": "adoptable_dogs_count" }
  ], session: session).first["adoptable_dogs_count"]
  puts adoptable_pets_count
end

La serie de comandos anterior:

Utiliza MongoClient() para establecer una conexión con la implementación de MongoDB.
Cambia a la base de datos pets.
Establece una sesión. El comando snapshot=True especifica, por lo que la sesión utiliza la preocupación de "snapshot" lectura.
Realiza estas acciones para cada colección en la base de datos pets:
- Utiliza para filtrar documentos donde $match el adoptable campo True es.
- Utiliza para devolver un recuento de los documentos $count filtrados.
- Incrementa la variable adoptablePetsCount con el recuento de la base de datos.
Imprime la variable adoptablePetsCount.

Todas las consultas de la sesión leen los datos tal como aparecieron en el mismo momento. Como resultado, el recuento final refleja una instantánea consistente de los datos.

Nota

Si la sesión dura más que el periodo de retención del historial de WiredTiger (300 segundos, por defecto), la consulta genera un SnapshotTooOld error. Para saber cómo configurar la retención de instantáneas y habilitar consultas de mayor duración, consulte Configurar la retención de instantáneas.

Leer desde un estado consistente de los datos de algún punto en el pasado

La lectura de la preocupación garantiza que su consulta lea los datos tal como aparecieron en un momento determinado en el pasado reciente."snapshot"

Una tienda de zapatos en línea tiene una colección sales que contiene datos de cada artículo vendido. Por ejemplo, un documento de la colección sales se ve así:

{
   "shoeType": "boot",
   "price": 30,
   "saleDate": ISODate("2022-02-02T06:01:17.171Z")
}

Cada medianoche, se ejecuta una consulta para ver cuántos pares de zapatos se vendieron ese día. La consulta de ventas diarias se ve así:

mongoc_client_session_t *cs = NULL;
mongoc_collection_t *sales_collection = NULL;
bson_error_t error;
mongoc_session_opt_t *session_opts;
bson_t *pipeline = NULL;
bson_t opts = BSON_INITIALIZER;
mongoc_cursor_t *cursor = NULL;
const bson_t *doc = NULL;
bool ok = true;
bson_iter_t iter;
int64_t total_sales = 0;
sales_collection = mongoc_client_get_collection(client, "retail", "sales");
/* seed 'retail.sales' with example data */
if (!retail_setup(sales_collection)) {
   goto cleanup;
}
/* start a snapshot session */
session_opts = mongoc_session_opts_new();
mongoc_session_opts_set_snapshot(session_opts, true);
cs = mongoc_client_start_session(client, session_opts, &error);
mongoc_session_opts_destroy(session_opts);
if (!cs) {
   MONGOC_ERROR("Could not start session: %s", error.message);
   goto cleanup;
}
if (!mongoc_client_session_append(cs, &opts, &error)) {
   MONGOC_ERROR("could not apply session options: %s", error.message);
   goto cleanup;
}
pipeline = BCON_NEW("pipeline",
                    "[",
                    "{",
                    "$match",
                    "{",
                    "$expr",
                    "{",
                    "$gt",
                    "[",
                    "$saleDate",
                    "{",
                    "$dateSubtract",
                    "{",
                    "startDate",
                    "$$NOW",
                    "unit",
                    BCON_UTF8("day"),
                    "amount",
                    BCON_INT64(1),
                    "}",
                    "}",
                    "]",
                    "}",
                    "}",
                    "}",
                    "{",
                    "$count",
                    BCON_UTF8("totalDailySales"),
                    "}",
                    "]");
cursor = mongoc_collection_aggregate(sales_collection, MONGOC_QUERY_NONE, pipeline, &opts, NULL);
bson_destroy(&opts);
ok = mongoc_cursor_next(cursor, &doc);
if (mongoc_cursor_error(cursor, &error)) {
   MONGOC_ERROR("could not get totalDailySales: %s", error.message);
   goto cleanup;
}
if (!ok) {
   MONGOC_ERROR("%s", "cursor has no results");
   goto cleanup;
}
ok = bson_iter_init_find(&iter, doc, "totalDailySales");
if (ok) {
   total_sales = bson_iter_as_int64(&iter);
} else {
   MONGOC_ERROR("%s", "missing key: 'totalDailySales'");
   goto cleanup;
}

using namespace mongocxx;
using bsoncxx::builder::basic::kvp;
using bsoncxx::builder::basic::make_array;
using bsoncxx::builder::basic::make_document;
auto opts = mongocxx::options::client_session{};
opts.snapshot(true);
auto session = client.start_session(opts);
auto db = client["retail"];
pipeline p;
p
    .match(make_document(kvp(
        "$expr",
        make_document(kvp(
            "$gt",
            make_array(
                "$saleDate", make_document(kvp("startDate", "$$NOW"), kvp("unit", "day"), kvp("amount", 1))))))))
    .count("totalDailySales");
auto cursor = db["sales"].aggregate(session, p);
auto doc = *cursor.begin();
auto total_daily_sales = doc.find("totalDailySales")->get_int32();

ctx := context.TODO()
sess, err := client.StartSession(options.Session().SetSnapshot(true))
if err != nil {
	return err
}
defer sess.EndSession(ctx)
var totalDailySales int32
err = mongo.WithSession(ctx, sess, func(ctx context.Context) error {
	// Count the total daily sales
	const totalDailySalesOutput = "totalDailySales"
	cursor, err := db.Collection("sales").Aggregate(ctx, mongo.Pipeline{
		bson.D{{
			"$match",
			bson.D{{
				"$expr",
				bson.D{{
					"$gt",
					bson.A{
						"$saleDate",
						bson.D{{
							"$dateSubtract",
							bson.D{
								{"startDate", "$$NOW"},
								{"unit", "day"},
								{"amount", 1},
							},
						}},
					},
				}},
			}},
		}},
		bson.D{{"$count", totalDailySalesOutput}},
	})
	if err != nil {
		return err
	}
	if !cursor.Next(ctx) {
		return fmt.Errorf("expected aggregate to return a document, but got none")
	}
	resp := cursor.Current.Lookup(totalDailySalesOutput)
	var ok bool
	totalDailySales, ok = resp.Int32OK()
	if !ok {
		return fmt.Errorf("failed to find int32 field %q in document %v", totalDailySalesOutput, cursor.Current)
	}
	return nil
})
if err != nil {
	return err
}

db = client.retail
async with await client.start_session(snapshot=True) as s:
    docs = await db.sales.aggregate(
        [
            {
                "$match": {
                    "$expr": {
                        "$gt": [
                            "$saleDate",
                            {
                                "$dateSubtract": {
                                    "startDate": "$$NOW",
                                    "unit": "day",
                                    "amount": 1,
                                }
                            },
                        ]
                    }
                }
            },
            {"$count": "totalDailySales"},
        ],
        session=s,
    ).to_list(None)
    total = docs[0]["totalDailySales"]
    print(total)

$salesCollection = $client->selectCollection('retail', 'sales');
$session = $client->startSession(['snapshot' => true]);
$totalDailySales = $salesCollection->aggregate(
    [
        [
            '$match' => [
                '$expr' => [
                    '$gt' => ['$saleDate', [
                        '$dateSubtract' => [
                            'startDate' => '$$NOW',
                            'unit' => 'day',
                            'amount' => 1,
                        ],
                    ],
                    ],
                ],
            ],
        ],
        ['$count' => 'totalDailySales'],
    ],
    ['session' => $session],
)->toArray()[0]->totalDailySales;

db = client.retail
with client.start_session(snapshot=True) as s:
    _ = (
        (
            db.sales.aggregate(
                [
                    {
                        "$match": {
                            "$expr": {
                                "$gt": [
                                    "$saleDate",
                                    {
                                        "$dateSubtract": {
                                            "startDate": "$$NOW",
                                            "unit": "day",
                                            "amount": 1,
                                        }
                                    },
                                ]
                            }
                        }
                    },
                    {"$count": "totalDailySales"},
                ],
                session=s,
            )
        ).next()
    )["totalDailySales"]

client = Mongo::Client.new(uri_string, database: "retail")
client.start_session(snapshot: true) do |session|
  total = client['sales'].aggregate([
    {
       "$match": {
          "$expr": {
             "$gt": [
                "$saleDate",
                {
                   "$dateSubtract": {
                      startDate: "$$NOW",
                      unit: "day",
                      amount: 1
                   }
                }
             ]
           }
        }
    },
    { "$count": "total_daily_sales" }
  ], session: session).first["total_daily_sales"]
end

La consulta anterior:

Utiliza $match con para especificar un filtro en $expr el saleDate campo.
- $expr permite el uso de expresiones de NOW agregación (como) en la $match etapa.
Utiliza el $gt operador y $dateSubtract la expresión para devolver documentos donde saleDate es mayor que un día antes del momento en que se ejecuta la consulta.
Utiliza para devolver un recuento de los documentos coincidentes. Este recuento se almacena en $count la totalDailySales variable.
Especifica "snapshot" la preocupación de lectura para garantizar que la consulta se lea desde un único punto en el tiempo.

La colección sales es bastante grande, por lo que esta consulta puede tardar unos minutos en ejecutarse. Como la tienda está en línea, las ventas pueden ocurrir a cualquier hora del día.

Por ejemplo, considere lo siguiente:

La consulta comienza a ejecutarse a las 12:00 AM.
Un cliente compra tres pares de zapatos a las 12:02 AM.
La consulta finaliza su ejecución a las 12:04 AM.

Si la consulta no utiliza la preocupación de lectura, las ventas que se producen entre el inicio y el final de la consulta pueden incluirse en el recuento, aunque no se hayan producido el día del informe. Esto podría generar informes inexactos, con algunas ventas contabilizadas dos "snapshot" veces.

Al especificar la preocupación de "snapshot" lectura, la consulta solo devuelve datos que estaban presentes en la base de datos en un momento poco antes de que la consulta comenzara a ejecutarse.

Nota

Si la consulta tarda más que el periodo de retención del historial de WiredTiger (300 segundos, por defecto), se produce un SnapshotTooOld error. Para saber cómo configurar la retención de instantáneas y habilitar consultas de mayor duración, consulte Configurar la retención de instantáneas.

Configurar la retención de instantáneas

De forma predeterminada, el motor de almacenamiento de WiredTiger conserva el historial durante 300 segundos. Puede usar una sesión con snapshot=true durante un total de 300 segundos desde la primera operación hasta la última. Si usa la sesión durante más tiempo, fallará con un SnapshotTooOld error. De igual forma, si consulta datos con la consulta de lectura y dura más "snapshot" de 300 segundos, la consulta fallará.

Si su consulta o sesión se ejecuta durante más de 300 segundos, considere aumentar el periodo de retención de la instantánea. Para ello, modifique el minSnapshotHistoryWindowInSeconds parámetro.

Por ejemplo, este comando establece el valor de minSnapshotHistoryWindowInSeconds en 600 segundos:

db.adminCommand( { setParameter: 1, minSnapshotHistoryWindowInSeconds: 600 } )

Importante

Para modificar para minSnapshotHistoryWindowInSeconds un clúster MongoDB Atlas, debe comunicarse con el soporte de Atlas.

Espacio en disco e historial

Incrementar el valor de minSnapshotHistoryWindowInSeconds incrementa el uso del disco porque el servidor debe mantener el historial de valores modificados anteriores dentro de la ventana de tiempo especificada. La cantidad de espacio en disco utilizado depende de la carga de trabajo, y las cargas de trabajo de mayor volumen requieren más espacio en disco.

Volver

Tiempos de espera

Update