Utilisation des flux de modifications avec Amazon DocumentDB

La fonctionnalité de flux de modifications d'Amazon DocumentDB (avec compatibilité avec MongoDB) fournit une séquence chronologique des événements de modification qui se produisent dans les collections de votre cluster. Vous pouvez lire des événements à partir d'un flux de modifications afin d'implémenter de nombreux cas d'utilisation différents, notamment les suivants :

• Notification de modification

• Full-text recherche avec Amazon OpenSearch Service (OpenSearch Service)

• Analyses avec Amazon Redshift

Les applications peuvent utiliser les flux de modifications pour souscrire à tous les changements de données dans des collections individuelles. Les événements de flux de modifications sont ordonnés au fur et à mesure qu'ils se produisent sur le cluster et sont stockés pendant trois heures (par défaut) après l'enregistrement de l'événement. La période de conservation peut être prolongée jusqu'à 7 jours à l'aide du change_stream_log_retention_duration paramètre. Pour modifier la période de rétention du flux de modifications, consultezModifier la durée de conservation du journal des flux de modifications.

Opérations prises en charge

Amazon DocumentDB prend en charge les opérations suivantes pour les flux de modifications :

• Tous les événements de changement sont pris en charge dans MongoDB db.collection.watch() db.watch() et dans client.watch() l'API.

• Recherche complète de documents pour les mises à jour.

• Étapes d'agrégation : $match $project$redact,, $addFields et$replaceRoot.

• Reprise d'un flux de modifications à partir d'un jeton de reprise

• Reprise d'un flux de modifications à partir d'un horodatage en utilisant startAtOperation (applicable à Amazon DocumentDB 4.0+)

Facturation

La fonctionnalité de flux de modifications d'Amazon DocumentDB est désactivée par défaut et n'entraîne aucun frais supplémentaire tant que la fonctionnalité n'est pas activée. L'utilisation de flux de modifications dans un cluster entraîne des coûts supplémentaires de lecture et d'écriture sur iOS et de stockage. Vous pouvez utiliser l'opération modifyChangeStreams API pour activer cette fonctionnalité pour votre cluster. Pour plus d'informations sur la tarification, consultez la rubrique Tarification d'Amazon DocumentDB.

Limitations

Les flux de modifications présentent les limites suivantes dans Amazon DocumentDB :

• Sur Amazon DocumentDB 3.6 et Amazon DocumentDB 4.0, les flux de modifications ne peuvent être ouverts qu'à partir d'une connexion à l'instance principale d'un cluster Amazon DocumentDB. La lecture à partir de flux de modifications sur une instance de réplication n'est pas prise en charge sur Amazon DocumentDB 3.6 et Amazon DocumentDB 4.0. Lorsque vous appelez l'opération d'API watch(), vous devez spécifier une préférence de lecture primary pour vous assurer que toutes les lectures sont dirigées vers l'instance principale (veuillez consulter la section Exemple).

• Sur Amazon DocumentDB 5.0, les flux de modifications peuvent être ouverts à la fois à partir de l'instance principale et des instances secondaires, y compris des clusters globaux. Vous pouvez spécifier une préférence de lecture secondaire pour rediriger les flux de modifications vers des instances secondaires. Consultez Utilisation de flux de modifications sur les instances secondaires les meilleures pratiques et limites supplémentaires.

• Les événements écrits dans un flux de modifications pour une collection sont disponibles pendant une durée maximale de 7 jours (la valeur par défaut est de 3 heures). Les données du flux de modifications sont supprimées après la fenêtre de durée de conservation du journal, même si aucune nouvelle modification n'est survenue.

• Une opération d'écriture longue en cours d'exécution sur une collection comme updateMany ou deleteMany peut bloquer temporairement l'écriture des événements de flux de modifications jusqu'à ce que l'opération d'écriture longue en cours d'exécution soit terminée.

• Amazon DocumentDB ne prend pas en charge le journal des opérations MongoDB ()oplog.

• Avec Amazon DocumentDB, vous devez activer explicitement les flux de modifications sur une collection donnée.

• Si la taille totale d'un événement de flux de modifications (y compris les données de modification et le document complet, le cas échéant) est supérieure à 16 MB, le client rencontre un échec de lecture sur les flux de modification.

• Le pilote Ruby n'est actuellement pas pris en charge lors de l'utilisation db.watch() et client.watch() avec Amazon DocumentDB 3.6.

• Le résultat de la updateDescription commande dans les flux de modifications est différent dans Amazon DocumentDB et dans MongoDB lorsque la valeur mise à jour du champ est la même que la précédente :

  • Amazon DocumentDB ne renvoie aucun champ dans la updateDescription sortie si le champ fourni est spécifié dans la $set commande et que sa valeur cible est déjà égale à la valeur source.

  • MongoDB renvoie le champ dans la sortie, même si la valeur spécifiée est égale à la valeur actuelle.

Activer les flux de modifications

Vous pouvez activer les flux de modifications Amazon DocumentDB pour toutes les collections d'une base de données donnée, ou uniquement pour certaines collections. Voici des exemples de la façon d'activer les flux de changement pour différents cas d'utilisation en utilisant le shell mongo. Les chaînes vides sont traitées comme des caractères génériques lors de la spécification des noms de base de données et de collections.

//Enable change streams for the collection "foo" in database "bar"
db.adminCommand({modifyChangeStreams: 1,
    database: "bar",
    collection: "foo", 
    enable: true});

//Disable change streams on collection "foo" in database "bar"
db.adminCommand({modifyChangeStreams: 1,
    database: "bar",
    collection: "foo", 
    enable: false});

//Enable change streams for all collections in database "bar"
db.adminCommand({modifyChangeStreams: 1,
    database: "bar",
    collection: "", 
    enable: true});

//Enable change streams for all collections in all databases in a cluster
db.adminCommand({modifyChangeStreams: 1,
    database: "",
    collection: "", 
    enable: true});

Les flux de modifications seront activés pour une collection si l'un des éléments suivants est vrai :

• La base de données et la collection sont explicitement activées.

• La base de données contenant la collection est activée.

• Toutes les bases de données sont activées.

Le fait de supprimer une collection d'une base de données ne désactive pas les flux de modification pour cette collection si la base de données parent a également activé les flux de modification ou si toutes les bases de données du cluster sont activées. Si une nouvelle collection est créée avec le même nom que la collection supprimée, les flux de modifications seront activés pour cette collection.

Vous pouvez répertorier tous les flux de modifications activés de votre cluster à l'aide de l'étape du pipeline d'agrégation $listChangeStreams. Toutes les étapes d'agrégation prises en charge par Amazon DocumentDB peuvent être utilisées dans le pipeline pour un traitement supplémentaire. Si une collection précédemment activée a été désactivée, elle n'apparaît pas dans la sortie $listChangeStreams.

//List all databases and collections with change streams enabled
cursor = new DBCommandCursor(db,
    db.runCommand(
        {aggregate: 1,
        pipeline: [{$listChangeStreams: 1}], 
        cursor:{}}));

//List of all databases and collections with change streams enabled 
{ "database" : "test", "collection" : "foo" } 
{ "database" : "bar", "collection" : "" }
{ "database" : "", "collection" : "" }

//Determine if the database “bar” or collection “bar.foo” have change streams enabled
cursor = new DBCommandCursor(db,
  db.runCommand(
      {aggregate: 1,
       pipeline: [{$listChangeStreams: 1},
                  {$match: {$or: [{database: "bar", collection: "foo"},
                                  {database: "bar", collection: ""},
                                  {database: "", collection: ""}]}}
                 ],
      cursor:{}}));

Exemple : utilisation de flux de modifications avec Python

Voici un exemple d'utilisation d'un flux de modification Amazon DocumentDB avec Python au niveau de la collection.

import os
import sys
from pymongo import MongoClient, ReadPreference
      
username = "DocumentDBusername"
password = <Insert your password> 

clusterendpoint = "DocumentDBClusterEndpoint”
client = MongoClient(clusterendpoint, username=username, password=password, tls='true', tlsCAFile='global-bundle.pem')

db = client['bar']
 
#While ‘Primary’ is the default read preference, here we give an example of
#how to specify the required read preference when reading the change streams
coll = db.get_collection('foo', read_preference=ReadPreference.PRIMARY)
#Create a stream object
stream = coll.watch()
#Write a new document to the collection to generate a change event
coll.insert_one({'x': 1})
#Read the next change event from the stream (if any)
print(stream.try_next())

"""
Expected Output:
{'_id': {'_data': '015daf94f600000002010000000200009025'},
'clusterTime': Timestamp(1571788022, 2),
'documentKey': {'_id': ObjectId('5daf94f6ea258751778163d6')},
'fullDocument': {'_id': ObjectId('5daf94f6ea258751778163d6'), 'x': 1},
'ns': {'coll': 'foo', 'db': 'bar'},
'operationType': 'insert'}
"""

#A subsequent attempt to read the next change event returns nothing, as there are no new changes
print(stream.try_next())

"""
Expected Output:
None
""" 
 
#Generate a new change event by updating a document
result = coll.update_one({'x': 1}, {'$set': {'x': 2}})
print(stream.try_next())

"""
Expected Output:
{'_id': {'_data': '015daf99d400000001010000000100009025'},
'clusterTime': Timestamp(1571789268, 1),
'documentKey': {'_id': ObjectId('5daf9502ea258751778163d7')},
'ns': {'coll': 'foo', 'db': 'bar'},
'operationType': 'update',
'updateDescription': {'removedFields': [], 'updatedFields': {'x': 2}}}
"""

Voici un exemple d'utilisation d'un flux de modifications Amazon DocumentDB avec Python au niveau de la base de données.

import os
import sys
from pymongo import MongoClient

username = "DocumentDBusername"
password = <Insert your password>
clusterendpoint = "DocumentDBClusterEndpoint”
client = MongoClient(clusterendpoint, username=username, password=password, tls='true', tlsCAFile='global-bundle.pem')

db = client['bar']
#Create a stream object
stream = db.watch()
coll = db.get_collection('foo')
#Write a new document to the collection foo to generate a change event
coll.insert_one({'x': 1})
 
#Read the next change event from the stream (if any)
print(stream.try_next())

"""
Expected Output:
{'_id': {'_data': '015daf94f600000002010000000200009025'},
'clusterTime': Timestamp(1571788022, 2),
'documentKey': {'_id': ObjectId('5daf94f6ea258751778163d6')},
'fullDocument': {'_id': ObjectId('5daf94f6ea258751778163d6'), 'x': 1},
'ns': {'coll': 'foo', 'db': 'bar'},
'operationType': 'insert'}
"""
#A subsequent attempt to read the next change event returns nothing, as there are no new changes
print(stream.try_next())

"""
Expected Output:
None
""" 
 
coll = db.get_collection('foo1')

#Write a new document to another collection to generate a change event
coll.insert_one({'x': 1})
print(stream.try_next())

"""
Expected Output: Since the change stream cursor was the database level you can see change events from different collections in the same database
{'_id': {'_data': '015daf94f600000002010000000200009025'},
'clusterTime': Timestamp(1571788022, 2),
'documentKey': {'_id': ObjectId('5daf94f6ea258751778163d6')},
'fullDocument': {'_id': ObjectId('5daf94f6ea258751778163d6'), 'x': 1},
'ns': {'coll': 'foo1', 'db': 'bar'},
'operationType': 'insert'}
"""

Recherche complète de documents

L'événement de modification de mise à jour n'inclut pas le document complet, mais uniquement la modification qui a été effectuée. Si votre cas d'utilisation nécessite le document complet affecté par une mise à jour, vous pouvez activer la recherche complète de documents lors de l'ouverture du flux.

Le document fullDocument d'un événement de flux de modification de mise à jour représente la version la plus récente du document mis à jour au moment de la recherche de document. Si des modifications se sont produites entre l'opération de mise à jour et la recherche fullDocument, le document fullDocument peut ne pas représenter l'état du document au moment de la mise à jour.

Pour créer un objet de flux avec la recherche de mise à jour activée, utilisez cet exemple :

stream = coll.watch(full_document='updateLookup')
 
#Generate a new change event by updating a document
result = coll.update_one({'x': 2}, {'$set': {'x': 3}})

stream.try_next()

La sortie de l'objet stream ressemblera à ceci :

{'_id': {'_data': '015daf9b7c00000001010000000100009025'},
'clusterTime': Timestamp(1571789692, 1),
'documentKey': {'_id': ObjectId('5daf9502ea258751778163d7')},
'fullDocument': {'_id': ObjectId('5daf9502ea258751778163d7'), 'x': 3},
'ns': {'coll': 'foo', 'db': 'bar'},
'operationType': 'update',
'updateDescription': {'removedFields': [], 'updatedFields': {'x': 3}}}

Reprise d'un flux de modifications

Vous pouvez reprendre un flux de modifications ultérieurement à l'aide d'un jeton de reprise, qui est égal au champ _iddu dernier document d'événement de modification récupéré.

import os
import sys
from pymongo import MongoClient

username = "DocumentDBusername"
password = <Insert your password> 
clusterendpoint = "DocumentDBClusterEndpoint”
client = MongoClient(clusterendpoint, username=username, password=password, tls='true', tlsCAFile='global-bundle.pem', retryWrites='false')

db = client['bar']
coll = db.get_collection('foo')
#Create a stream object
stream = db.watch()
coll.update_one({'x': 1}, {'$set': {'x': 4}})
event = stream.try_next()
token = event['_id']
print(token)

"""
Output: This is the resume token that we will later us to resume the change stream
{'_data': '015daf9c5b00000001010000000100009025'}
"""
#Python provides a nice shortcut for getting a stream’s resume token
print(stream.resume_token)

"""
Output
{'_data': '015daf9c5b00000001010000000100009025'}
"""
#Generate a new change event by updating a document
result = coll.update_one({'x': 4}, {'$set': {'x': 5}})
#Generate another change event by inserting a document
result = coll.insert_one({'y': 5})
#Open a stream starting after the selected resume token
stream = db.watch(full_document='updateLookup', resume_after=token)
#Our first change event is the update with the specified _id
print(stream.try_next())

"""
#Output: Since we are resuming the change stream from the resume token, we will see all events after the first update operation. In our case, the change stream will resume from the update operation {x:5}

{'_id': {'_data': '015f7e8f0c000000060100000006000fe038'}, 
'operationType': 'update', 
'clusterTime': Timestamp(1602129676, 6), 
'ns': {'db': 'bar', 'coll': 'foo'}, 
'documentKey': {'_id': ObjectId('5f7e8f0ac423bafbfd9adba2')}, 
'fullDocument': {'_id': ObjectId('5f7e8f0ac423bafbfd9adba2'), 'x': 5}, 
'updateDescription': {'updatedFields': {'x': 5}, 'removedFields': []}}
"""
#Followed by the insert
print(stream.try_next())

"""
#Output:
{'_id': {'_data': '015f7e8f0c000000070100000007000fe038'}, 
'operationType': 'insert', 
'clusterTime': Timestamp(1602129676, 7), 
'ns': {'db': 'bar', 'coll': 'foo'}, 
'documentKey': {'_id': ObjectId('5f7e8f0cbf8c233ed577eb94')}, 
'fullDocument': {'_id': ObjectId('5f7e8f0cbf8c233ed577eb94'), 'y': 5}}
"""

Reprise d'un flux de modifications avec start AtOperationTime

Vous pouvez reprendre un flux de modifications ultérieurement à partir d'un horodatage donné en utilisantstartAtOperationTime.

Note

La possibilité d'utilisation startAtOperationTime est disponible dans Amazon DocumentDB 4.0+. Lors de l'utilisationstartAtOperationTime, le curseur du flux de modifications ne renvoie que les modifications survenues pendant ou après l'horodatage spécifié. Les resumeAfter commandes startAtOperationTime et s'excluent mutuellement et ne peuvent donc pas être utilisées ensemble.

import os
import sys
from pymongo import MongoClient

username = "DocumentDBusername"
password = <Insert your password> 
clusterendpoint = "DocumentDBClusterEndpoint”
client = MongoClient(clusterendpoint, username=username, password=password, tls='true', tlsCAFile='rds-root-ca-2020.pem',retryWrites='false')
db = client['bar']
coll = db.get_collection('foo')
#Create a stream object
stream = db.watch()
coll.update_one({'x': 1}, {'$set': {'x': 4}})
event = stream.try_next()
timestamp = event['clusterTime']
print(timestamp)
"""
Output
Timestamp(1602129114, 4)
"""
#Generate a new change event by updating a document
result = coll.update_one({'x': 4}, {'$set': {'x': 5}})
result = coll.insert_one({'y': 5})
#Generate another change event by inserting a document
#Open a stream starting after specified time stamp

stream = db.watch(start_at_operation_time=timestamp)
print(stream.try_next())

"""
#Output: Since we are resuming the change stream at the time stamp of our first update operation (x:4), the change stream cursor will point to that event
{'_id': {'_data': '015f7e941a000000030100000003000fe038'}, 
'operationType': 'update', 
'clusterTime': Timestamp(1602130970, 3), 
'ns': {'db': 'bar', 'coll': 'foo'}, 
'documentKey': {'_id': ObjectId('5f7e9417c423bafbfd9adbb1')}, 
'updateDescription': {'updatedFields': {'x': 4}, 'removedFields': []}}
"""

print(stream.try_next())
"""
#Output: The second event will be the subsequent update operation (x:5)
{'_id': {'_data': '015f7e9502000000050100000005000fe038'}, 
'operationType': 'update', 
'clusterTime': Timestamp(1602131202, 5),
'ns': {'db': 'bar', 'coll': 'foo'}, 
'documentKey': {'_id': ObjectId('5f7e94ffc423bafbfd9adbb2')}, 
'updateDescription': {'updatedFields': {'x': 5}, 'removedFields': []}}
"""

print(stream.try_next())

"""
#Output: And finally the last event will be the insert operation (y:5)
{'_id': {'_data': '015f7e9502000000060100000006000fe038'}, 
'operationType': 'insert', 
'clusterTime': Timestamp(1602131202, 6), 
'ns': {'db': 'bar', 'coll': 'foo'}, 
'documentKey': {'_id': ObjectId('5f7e95025c4a569e0f6dde92')}, 
'fullDocument': {'_id': ObjectId('5f7e95025c4a569e0f6dde92'), 'y': 5}}
"""

Reprise d'un flux de modifications avec publication BatchResumeToken

Le flux de modification Amazon DocumentDB renvoie désormais un champ supplémentaire appelépostBatchResumeToken. Ce champ est renvoyé par la $changestream commande et la getMore commande.

Exemple de $changestream commande en Python :

db.command({"aggregate": "sales", "pipeline": [{ "$changeStream": {}}], "cursor": {"batchSize": 1}})

Sortie attendue :

cursor" : {
   "firstBatch" : [ ],
   "postBatchResumeToken" : {"_data" : "0167c8cbe60000000004"},
   "id" : NumberLong("9660788144470"),
   "ns" : "test.sales"
}

Exemple de getMore commande en Python :

db.command({"getMore": NumberLong(<cursor id>), "collection": "sales", "batchSize": 1 })

Sortie attendue

cursor" : {
   "nextBatch" : [ ],
   "postBatchResumeToken" : {"_data" : "0167c8cbe60000000004"},
   "id" : NumberLong("9660788144470"),
   "ns" : "test.sales"
}

Le postBatchResumeToken champ peut être utilisé pour ouvrir de nouveaux curseurs de flux de modifications dans le resumeAfter champ, de la même manière que le jeton de reprise est utilisé.

Ouvrez un flux commençant après la sélection postBatchResumeToken :

post_batch_resume_token = output['cursor']['postBatchResumeToken']
stream = db.watch(full_document='updateLookup', resume_after=post_batch_resume_token)

Contrairement à un jeton de reprise classique qui correspond toujours à une entrée du journal des opérations (oplog) reflétant un événement réel, il postBatchResumeToken correspond à une entrée oplog que le flux de modifications a parcourue sur le serveur, ce qui n'est pas nécessairement une modification correspondante.

Toute tentative de reprise avec un ancien jeton de CV normal forcera la base de données à scanner toutes les entrées du journal entre l'horodatage spécifié et l'heure actuelle. Cela peut générer de nombreuses requêtes en interne, chaque sous-requête étant analysée pendant une courte période. Cela provoquera un pic d'utilisation du processeur et dégradera les performances de la base de données. Reprise avec les derniers postBatchResumeToken sauts de l'analyse des entrées de journal non concordantes.

Transactions dans les flux de changement

Les événements du flux de modifications ne contiendront pas d'événements provenant de transactions and/or abandonnées non validées. Par exemple, si vous démarrez une INSERT transaction avec une UPDATE opération et si votre INSERT opération réussit mais échoue, la transaction sera annulée. UPDATE Cette transaction ayant été annulée, votre flux de modifications ne contiendra aucun événement pour cette transaction.

Modifier la durée de conservation du journal des flux de modifications

Vous pouvez modifier la durée de conservation du journal des flux de modifications pour qu'elle soit comprise entre 1 heure et 7 jours à l'aide du Console de gestion AWS ou du AWS CLI.

Using the Console de gestion AWS

Pour modifier la durée de conservation du journal du flux de modifications

1. Connectez-vous au et ouvrez Console de gestion AWS la console Amazon DocumentDB à https://console.aws.amazon.com/docdb l'adresse.

2. Dans le panneau de navigation, choisissez Groupes de paramètres.

   Astuce

   Si vous ne voyez pas le volet de navigation sur le côté gauche de votre écran, choisissez l'icône de menu () dans le coin supérieur gauche de la page.

3. Dans le volet Parameter groups (Groupes de paramètres), choisissez le groupe de paramètres de cluster associé à votre cluster. Pour identifier le groupe de paramètres de cluster associé à votre cluster, veuillez consulter Déterminer le groupe de paramètres d'un cluster Amazon DocumentDB.

4. La page qui s’affiche contient les paramètres et leurs détails correspondants pour votre groupe de paramètres de cluster. Sélectionnez le paramètre change_stream_log_retention_duration.

5. En haut à droite de la page, choisissez Edit (Modifier) pour modifier la valeur du paramètre. Le change_stream_log_retention_duration paramètre peut être modifié pour être compris entre 1 heure et 7 jours.

6. Effectuez votre modification, puis choisissez Modify cluster parameter (Modifier le paramètre de cluster) pour enregistrer les modifications. Pour ignorer vos modifications, choisissez Annuler.

Using the AWS CLI

Pour modifier le paramètre change_stream_log_retention_duration de votre groupe de paramètres de cluster, utilisez l'opération modify-db-cluster-parameter-group avec les paramètres suivants :

• --db-cluster-parameter-group-name — Obligatoire. Nom du groupe de paramètres de cluster que vous voulez modifier. Pour identifier le groupe de paramètres de cluster associé à votre cluster, veuillez consulter Déterminer le groupe de paramètres d'un cluster Amazon DocumentDB.

• --parameters — Obligatoire. Paramètre que vous modifiez. Chaque saisie de paramètre doit inclure ce qui suit :

  • ParameterName— Le nom du paramètre que vous êtes en train de modifier. Dans ce cas, il s’agit de change_stream_log_retention_duration

  • ParameterValue— La nouvelle valeur pour ce paramètre.

  • ApplyMethod— Comment vous souhaitez que les modifications apportées à ce paramètre soient appliquées. Les valeurs autorisées sont immediate et pending-reboot.

    Note

    Les paramètres avec le ApplyType de static doivent avoir une ApplyMethod de pending-reboot.

1. Pour modifier les valeurs du paramètre change_stream_log_retention_duration, exécutez la commande suivante et remplacez parameter-value par la valeur que vous souhaitez attribuer au paramètre.

   Pour Linux, macOS ou Unix :

   aws docdb modify-db-cluster-parameter-group \
       --db-cluster-parameter-group-name sample-parameter-group \
       --parameters "ParameterName=change_stream_log_retention_duration,ParameterValue=<parameter-value>,ApplyMethod=immediate"

   Pour Windows :

   aws docdb modify-db-cluster-parameter-group ^
       --db-cluster-parameter-group-name sample-parameter-group ^
       --parameters "ParameterName=change_stream_log_retention_duration,ParameterValue=<parameter-value>,ApplyMethod=immediate"

   La sortie de cette opération ressemble à ceci (format JSON).

   {
       "DBClusterParameterGroupName": "sample-parameter-group"
   }

2. Patientez au moins 5 minutes.

3. Répertoriez les valeurs de paramètre de sample-parameter-group pour vous assurer que vos modifications ont été prises en compte.

   Pour Linux, macOS ou Unix :

   aws docdb describe-db-cluster-parameters \
       --db-cluster-parameter-group-name sample-parameter-group

   Pour Windows :

   aws docdb describe-db-cluster-parameters ^
       --db-cluster-parameter-group-name sample-parameter-group

   La sortie de cette opération ressemble à ceci (format JSON).

   {
       "Parameters": [
           {
               "ParameterName": "audit_logs",
               "ParameterValue": "disabled",
               "Description": "Enables auditing on cluster.",
               "Source": "system",
               "ApplyType": "dynamic",
               "DataType": "string",
               "AllowedValues": "enabled,disabled",
               "IsModifiable": true,
               "ApplyMethod": "pending-reboot"
           },
           {
               "ParameterName": "change_stream_log_retention_duration",
               "ParameterValue": "12345",
               "Description": "Duration of time in seconds that the change stream log is retained and can be consumed.",
               "Source": "user",
               "ApplyType": "dynamic",
               "DataType": "integer",
               "AllowedValues": "3600-86400",
               "IsModifiable": true,
               "ApplyMethod": "immediate"
           }
       ]
   }

Note

La conservation des journaux des flux de modifications ne supprimera pas les journaux antérieurs à la change_stream_log_retention_duration valeur configurée tant que la taille du journal n'est pas supérieure à (>) 51 200 Mo.

Utilisation de flux de modifications sur les instances secondaires

Pour commencer à utiliser le flux de modifications sur les instances secondaires, ouvrez le curseur du flux de modifications avec readPreference comme secondaire.

Vous pouvez ouvrir un curseur de flux de modifications pour surveiller les événements de changement sur une collection spécifique ou sur toutes les collections d'un cluster ou d'une base de données. Vous pouvez ouvrir un curseur de flux de modifications sur n'importe quelle instance Amazon DocumentDB et récupérer des documents de flux de modifications à la fois depuis les instances Writer et Reader. Vous pouvez partager des jetons de flux de modifications (tels que resumeToken oustartOperationTime) entre différents curseurs de flux de modifications ouverts sur une instance de rédaction et de lecture.

Exemple

import os
import sys
from pymongo import MongoClient, ReadPreference
      
username = "DocumentDBusername"
password = <Your password> 

clusterendpoint = "DocumentDBClusterEndpoint"

client = MongoClient(clusterendpoint, username=username, password=password, tls='true', tlsCAFile='global-bundle.pem')

db = client['bar']
 
# Make sure to use SECONDARY to redirect cursor reads from secondary instances
coll = db.get_collection('foo', read_preference=ReadPreference.SECONDARY)

# Create a stream object on RO. The token needs to generated from PRIMARY.
stream = coll.watch(resumeAfter=token)

for event in stream:
   print(event)

Directives et limites pour les flux de modifications sur les instances secondaires

• Les événements du flux de modifications doivent être répliqués de l'instance principale vers les instances secondaires. Vous pouvez surveiller le décalage à partir de la DBInstanceReplicaLag métrique d'Amazon CloudWatch.

• Les horodatages des instances secondaires peuvent ne pas toujours être synchronisés avec ceux de l'instance principale. Dans ce cas, attendez-vous à des retards dans l'horodatage de l'instance secondaire afin qu'elle puisse rattraper son retard. Il est recommandé d'utiliser startAtOperationTime ou resumeToken pour démarrer la surveillance sur l'instance secondaire.

• Le débit des instances secondaires peut être inférieur à celui de l'instance principale si la taille de votre document est importantefullDocumentLookup, et si la charge de travail d'écriture simultanée sur l'instance principale est élevée. Il est recommandé de surveiller le taux d'accès de votre cache tampon sur le secondaire et de vous assurer qu'il est élevé.