Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Utilisation de flux de modifications avec Amazon DocumentDB

La fonctionnalité de flux de modifications d'Amazon DocumentDB (compatible 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 :

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 en utilisant le change_stream_log_retention_duration paramètre. Pour modifier la période de conservation du flux de modifications, consultez la section Modification de la durée de conservation du journal du flux de modifications.

Opérations prises en charge

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

Facturation

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

Limites

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

Activer les flux de changement

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 :

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 modifications 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 de documents complets

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 startAtOperationTime

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

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

Transactions dans les flux de changement

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

Modification de la durée de conservation du journal du 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 AWS Management Console ou du AWS CLI.

Using the AWS Management Console

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

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

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 de ce paramètre.

  • ApplyMethod— La manière dont vous souhaitez que les modifications soient appliquées à ce paramètre. 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"

   Le résultat de cette opération ressemble à ce qui suit (JSONformat).

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

   Le résultat de cette opération ressemble à ce qui suit (JSONformat).

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

Utilisation des 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 readPreference en le désignant comme secondaire.

Vous pouvez ouvrir un curseur de flux de modifications pour surveiller les événements de modification concernant une collection spécifique ou 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 les documents du flux de modifications à partir des instances de rédacteur et de lecteur. 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édacteur et de lecteur.

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 relatives aux flux de modifications sur les instances secondaires