Partager via

Prise en main de la capture des changements de données dans le magasin analytique pour Azure Cosmos DB

  • Article

S’APPLIQUE À : NoSQL MongoDB

Utilisez la capture des changements de données (CDC, Change data capture) dans le magasin analytique Azure Cosmos DB comme source pour Azure Data Factory ou Azure Synapse Analytics pour capturer des modifications spécifiques à vos données.

Notes

Veuillez noter que l’interface de service lié pour l’API Azure Cosmos DB for MongoDB n’est pas encore disponible sur Dataflow. Toutefois, vous pouvez utiliser le point de terminaison de document de votre compte avec l’interface de service lié « Azure Cosmos DB for NoSQL » comme solution de contournement jusqu’à ce que le service lié Mongo soit pris en charge. Sur un service lié NoSQL, choisissez « Entrer manuellement » pour fournir les informations du compte Cosmos DB et utiliser le point de terminaison de document du compte (par exemple, https://[your-database-account-uri].documents.azure.com:443/) au lieu du point de terminaison MongoDB (par exemple : mongodb://[your-database-account-uri].mongo.cosmos.azure.com:10255/)

Prérequis

Activer le magasin analytique

Tout d’abord, activez Azure Synapse Link au niveau du compte, puis activez le magasin analytique pour les conteneurs qui conviennent à votre charge de travail.

  1. Activer Azure Synapse Link : Activer Azure Synapse Link pour un compte Azure Cosmos DB

  2. Activez le magasin analytique pour votre ou vos conteneurs :

    Option Guide
    Activation pour un nouveau conteneur spécifique Activation d’Azure Synapse Link pour vos nouveaux conteneurs
    Activation pour un conteneur existant spécifique Activation d’Azure Synapse Link pour vos conteneurs existants

Créer une ressource Azure cible à l’aide de flux de données

La fonctionnalité de capture des changements de données du magasin analytique est disponible via la fonctionnalité de flux de données d’Azure Data Factory ou d’Azure Synapse Analytics. Pour ce guide, utilisez Azure Data Factory.

Important

Vous pouvez aussi utiliser Azure Synapse Analytics. D’abord, créez un espace de travail Azure Synapse si vous n’en avez pas déjà un. Dans l’espace de travail nouvellement créé, sélectionnez l’onglet Développer, sélectionnez Ajouter une nouvelle ressource, puis sélectionnez Flux de données.

  1. Créez un Azure Data Factory si vous n’en avez pas déjà un.

    Conseil

    Si possible, créez la fabrique de données dans la même région que votre compte Azure Cosmos DB.

  2. Lancez la fabrique de données nouvellement créée.

  3. Dans la fabrique de données, sélectionnez l’onglet Flux de données, puis sélectionnez Nouveau flux de données.

  4. Donnez un nom unique au flux de données nouvellement créé. Dans cet exemple, le flux de données se nomme cosmoscdc.

    Capture d’écran d’un nouveau flux de données nommé cosmoscdc.

Configurer les paramètres sources pour le conteneur de magasin analytique

Maintenant, créez et configurez une source pour transmettre des données à partir du magasin analytique du compte Azure Cosmos DB.

  1. Sélectionnez Ajouter une source.

    Capture d’écran de l’option de menu ajouter une source.

  2. Dans le champ Nom du flux de sortie, entrez cosmos.

    Capture d’écran de la dénomination de la source cosmos nouvellement créée.

  3. Dans la section Type de source, sélectionnez Inline.

    Capture d’écran de la sélection du type de source inline.

  4. Dans le champ Jeu de données, sélectionnez Azure - Azure Cosmos DB for NoSQL.

    Capture d’écran de la sélection d’Azure Cosmos DB for NoSQL comme type de jeu de données.

  5. Créez un nouveau service lié pour votre compte nommé cosmoslinkedservice. Sélectionnez votre compte Azure Cosmos DB for NoSQL existant dans la fenêtre de dialogue contextuelle Nouveau service lié, puis sélectionnez Ok. Dans cet exemple, nous sélectionnons un compte Azure Cosmos DB for NoSQL préexistant nommé msdocs-cosmos-source et une base de données nommée cosmicworks.

    Capture d’écran de la fenêtre de dialogue Nouveau service lié avec un compte Azure Cosmos DB sélectionné.

  6. Sélectionnez Analytique pour le type de magasin.

    Capture d’écran de l’option analytique sélectionnée pour un service lié.

  7. Sélectionnez l’onglet Options de source.

  8. Dans Options de source, sélectionnez votre conteneur cible et activez le Débogage de flux de données. Dans cet exemple, le conteneur est nommé products.

    Capture d’écran d’un conteneur source avec des produits nommés sélectionnés.

  9. Sélectionnez Débogage de flux de données. Dans la fenêtre de dialogue contextuelle Activer le débogage de flux de données, conservez les options par défaut, puis sélectionnez Ok.

    Capture d’écran de l’option de bascule pour activer le débogage de flux de données.

  10. L’onglet Options de source contient également d’autres options que vous pouvez activer si vous le souhaitez. Ce tableau décrit ces options :

Option Description
Capturer les mises à jour intermédiaires Activez cette option si vous souhaitez capturer l’historique des changements apportés aux éléments, y compris les modifications intermédiaires entre les lectures de capture des changements de données.
Capturer les suppressions Activez cette option pour capturer les enregistrements supprimés par l’utilisateur et les appliquer sur le récepteur. Les suppressions ne peuvent pas être appliquées sur les récepteurs Azure Data Explorer et Azure Cosmos DB.
Capturer les TTL du magasin transactionnel Activez cette option pour capturer les enregistrements supprimés TTL (time-to-live) du magasin transactionnel Azure Cosmos DB et les appliquer sur le récepteur. Les suppressions de TTL ne peuvent pas être appliquées sur les récepteurs Azure Data Explorer et Azure Cosmos DB.
Taille de lot en octets Ce paramètre est en réalité en gigaoctets. Spécifiez la taille en gigaoctets si vous souhaitez traiter par lot les flux de capture des changements de données
Configurations supplémentaires Configurations supplémentaires du magasin analytique Azure Cosmos DB et leurs valeurs. (Exemple : spark.cosmos.allowWhiteSpaceInFieldNames -> true)

Utilisation des options sources

Lorsque vous consultez les options Capture intermediate updates, Capture Deltes et Capture Transactional store TTLs, votre processus CDC crée et remplit le champ __usr_opTypedans le récepteur avec les valeurs suivantes :

Valeur Description Option
1 UPDATE Capturer des mises à jour intermédiaires
2 INSERT Il n’existe pas d’option pour les insertions, elle est activée par défaut
3 USER_DELETE Capturer les suppressions
4 TTL_DELETE Capturer les TTL du magasin transactionnel

Si vous devez différencier les enregistrements supprimés TTL des documents supprimés par les utilisateurs ou les applications, vous devez cocher les options Capture intermediate updates et Capture Transactional store TTLs. Ensuite, vous devez adapter vos processus ou applications ou requêtes CDC pour utiliser __usr_opType en fonction des besoins de votre entreprise.

Conseil

Si les consommateurs en aval doivent rétablir l’ordre des mises à jour avec l’option « capturer les mises à jour intermédiaires » activée, le champ du timestamp système _ts peut être utilisé comme champ de commande.

Créer et configurer des paramètres de récepteur pour les opérations de mise à jour et de suppression

Créez d’abord un récepteur Stockage Blob Azure simple, puis configurez le récepteur pour uniquement filtrer les données sur des opérations spécifiques.

  1. Créez un compte et un conteneur Stockage Blob Azure si vous n’en avez pas déjà un. Pour les exemples suivants, nous allons utiliser un compte nommé msdocsblobstorage et un conteneur nommé output.

    Conseil

    Si possible, créez un compte de stockage dans la même région que votre compte Azure Cosmos DB.

  2. Retournez dans Azure Data Factory et créez un nouveau récepteur pour la capture des changements de données à partir de votre source cosmos.

    Capture d’écran de l’ajout d’un nouveau récepteur connecté à la source existante.

  3. Donnez au récepteur un nom unique. Dans cet exemple, le récepteur est nommé storage.

    Capture d’écran de la dénomination du stockage de récepteur nouvellement créé.

  4. Dans la section Type de récepteur, sélectionnez Inline. Dans le champ Jeu de données, sélectionnez Delta.

    Capture d’écran de la sélection et du type de jeu de données Inline Delta pour le récepteur.

  5. Créez un service lié nommé storagelinkedservice pour votre compte à l’aide du Stockage Blob Azure. Sélectionnez votre compte Stockage Blob Azure existant dans la fenêtre de dialogue contextuelle Nouveau service lié, puis sélectionnez Ok. Dans cet exemple, nous sélectionnons un compte Stockage Blob Azure préexistant nommé msdocsblobstorage.

    Capture d’écran des options de type de service pour un nouveau service lié Delta.

    Capture d’écran de la fenêtre de dialogue Nouveau service lié avec un compte Stockage Blob Azure sélectionné.

  6. Sélectionnez l’onglet Settings (Paramètres).

  7. Dans Paramètres, définissez le Chemin du dossier sur le nom du conteneur blob. Dans cet exemple, le nom du conteneur est output.

    Capture d’écran de l’ensemble de sortie nommé du conteneur blob comme cible du récepteur.

  8. Repérez la section Méthode de mise à jour et modifiez les sélections pour autoriser uniquement les opérations de suppression et de mise à jour. Spécifiez également les Colonnes clés en tant que Liste de colonnes en utilisant le champ {_rid} comme identificateur unique.

    Capture d’écran des méthodes de mise à jour et des colonnes clés spécifiées pour le récepteur.

  9. Sélectionnez Valider pour assurer que vous n’avez commis aucune erreur ou omission. Ensuite, sélectionnez Publier pour publier le flux de données.

    Capture d’écran de l’option permettant de valider puis de publier le flux de données actuel.

Planifier l’exécution d’une capture des changements de données

Une fois qu’un flux de données a été publié, vous pouvez ajouter un nouveau pipeline pour déplacer et transformer vos données.

  1. Créer un pipeline Donnez un nom unique au pipeline. Dans cet exemple, le pipeline est nommé cosmoscdcpipeline.

    Capture d’écran de la nouvelle option de pipeline dans la section des ressources.

  2. Dans la section Activités, développez l’option Déplacer et transformer, puis sélectionnez Flux de données.

    Capture d’écran de l’option d’activité de flux de données dans la section des activités.

  3. Donnez un nom unique à l’activité de flux de données. Dans cet exemple, l’activité est nommée cosmoscdcactivity.

  4. Sous l’onglet Paramètres, sélectionnez le flux de données nommé cosmoscdc que vous avez précédemment créé dans ce guide. Ensuite, sélectionnez une taille de calcul basée sur le volume de données et sur la latence requise pour votre charge de travail.

    Capture d’écran des paramètres de configuration à la fois pour le flux de données et la taille de calcul de l’activité.

    Conseil

    Pour les tailles de données incrémentielles supérieures à 100 Go, nous recommandons la taille Personnalisée avec 32 cœurs (+16 cœurs pilotes).

  5. Sélectionnez Ajouter un déclencheur. Planifiez l’exécution de ce pipeline à une cadence logique pour votre charge de travail. Dans cet exemple, le pipeline est configuré pour une exécution toutes les cinq minutes.

    Capture d’écran du bouton d’ajout de déclencheur pour un nouveau pipeline.

    Capture d’écran d’une configuration de déclencheur basée sur une planification qui commence en 2023 et qui s’exécute toutes les cinq minutes.

    Notes

    La fenêtre de périodicité minimale pour les exécutions de capture des changements de données est d’une minute.

  6. Sélectionnez Valider pour assurer que vous n’avez commis aucune erreur ou omission. Sélectionnez Publier pour publier le pipeline.

  7. Observez les données placées dans le conteneur Stockage Blob Azure en tant que sortie du flux de données à l’aide de la capture des changements de données du magasin analytique Azure Cosmos DB.

    Capture d’écran des fichiers de sortie du pipeline dans le conteneur Stockage Blob Azure.

    Notes

    Le temps de démarrage initial du cluster peut prendre jusqu’à trois minutes. Pour éviter un temps de démarrage du cluster dans les exécutions de capture des changements de données suivantes, configurez la valeur TTL du cluster de flux de données. Pour plus d’informations sur le runtime d’intégration et le TTL, consultez Runtime d’intégration dans Azure Data Factory.

Travaux simultanés

La taille du lot dans les options sources, ou les situations avec le récepteur lent à ingérer le flux de modifications, peuvent entraîner l’exécution simultanée de plusieurs travaux. Pour éviter cette situation, définissez l’option Concurrence sur 1 dans les paramètres du pipeline pour vous assurer que les nouvelles exécutions ne sont pas déclenchées jusqu’à la fin de l’exécution actuelle.

Étapes suivantes