Datastream organise les données répliquées selon une hiérarchie à trois niveaux : flux, objets et événements. Un flux connecte une source de données à une destination, tandis qu'un objet représente une entité source spécifique dans ce flux, telle qu'une table de base de données. Un événement capture une seule modification de données d'un objet, comme une insertion dans une base de données.
Chaque événement inclut des données et des métadonnées qui identifient son flux et son objet sources. Vous pouvez utiliser ces métadonnées pour suivre l'origine des modifications, supprimer les enregistrements en double dans votre destination et maintenir l'ordre entre les consommateurs en aval.
À propos des événements
Chaque événement se compose de trois types de données :
- Données d'événement : représentent la modification des données elles-mêmes à partir de l'objet provenant de la source du flux. Chaque événement contient l'intégralité de la ligne qui a été modifiée.
- Métadonnées génériques : ces métadonnées apparaissent sur chaque événement généré par Datastream, qui est utilisé pour des actions telles que la suppression des données en double dans la destination.
- Métadonnées spécifiques à la source : ces métadonnées apparaissent sur chaque événement généré par une source de flux spécifique. Ces métadonnées varient selon la source.
Données d'événement
Les données d'événement correspondent à la charge utile de chaque modification provenant d'un objet donné issu d'une source de flux.
Les événements sont au format Avro ou JSON.
Lorsque vous utilisez le format Avro, l'événement contient l'index et la valeur de chaque colonne. À l'aide de l'index de colonne, le nom de la colonne et le type unifié peuvent être récupérés à partir du schéma dans l'en-tête Avro.
Lorsque vous utilisez le format JSON, l'événement contient le nom et la valeur de chaque colonne.
Les métadonnées d'événement peuvent servir à collecter des informations sur l'origine de l'événement, et à supprimer les données en double dans la destination et les événements de commande par le consommateur en aval.
Les tableaux suivants répertorient et décrivent les champs et les types de données des métadonnées d'événement génériques et spécifiques à la source.
Métadonnées génériques
Ces métadonnées sont cohérentes pour tous les types de flux.
| Champ | Type Avro | Type JSON | Description |
|---|---|---|---|
stream_name |
chaîne | chaîne | Nom unique du flux, tel que défini au moment de la création. |
read_method |
chaîne | chaîne | Indique si les données ont été lues à partir de la source à l'aide d'une méthode de capture de données modifiées (CDC, Change Data Capture) dans le cadre d'un remplissage d'historique ou dans le cadre d'une tâche de complément créée lors d'un rollback d'une transaction pendant la réplication CDC. Les valeurs possibles sont les suivantes :
|
object |
chaîne | chaîne | Nom utilisé pour regrouper différents types d'événements, généralement le nom de la table ou de l'objet dans la source. |
schema_key |
chaîne | string | Identifiant unique du schéma unifié de l'événement. |
uuid |
string | chaîne | Identifiant unique de l'événement généré par Datastream. |
read_timestamp |
timestamp-millis | chaîne | Horodatage (UTC) de la lecture de l'enregistrement par Datastream (horodatage de l'epoch en millisecondes). |
source_timestamp |
timestamp-millis | chaîne | Horodatage (UTC) de la modification de l'enregistrement dans la source (horodatage de l'epoch en millisecondes). |
sort_keys |
{"type": "array", "items": ["string", "long"]} |
tableau | Tableau des valeurs pouvant être utilisées pour trier les événements dans l'ordre dans lequel ils se sont produits. |
Métadonnées spécifiques à la source
Ces métadonnées sont associées aux événements CDC et de backfill d'une base de données source. Pour afficher ces métadonnées, sélectionnez une source dans le menu déroulant ci-dessous.
| Source | Champ | Type Avro | Type JSON | Description |
|---|---|---|---|---|
| MySQL | log_file |
chaîne | string | Fichier journal à partir duquel Datastream extrait les événements dans la réplication CDC lors de l'utilisation de la réplication basée sur binlog. |
| MySQL | log_position |
long | long | Position du journal (décalage) dans le journal binaire MySQL lors de l'utilisation de la réplication basée sur binlog. |
| MySQL | binlog_file |
chaîne | string | Fichier journal binaire à partir duquel Datastream extrait les événements dans la réplication CDC lors de l'utilisation de la réplication basée sur GTID. |
| MySQL | binlog_position |
long | long | Position (décalage) dans le journal binaire MySQL lors de l'utilisation de la réplication basée sur GTID. |
| MySQL | gtid |
chaîne | string | Identifiant de transaction global (GTID) qui identifie de manière unique la transaction pour l'événement lors de l'utilisation de la réplication basée sur le GTID. |
| MySQL | datastream_master_server_uuid |
chaîne | string | UUID de l'instance de serveur source MySQL à partir de laquelle Datastream lit les événements lors de l'utilisation de la réplication basée sur GTID. |
| MySQL | datastream_master_server_id |
long | long | ID du serveur de l'instance de serveur source MySQL à partir de laquelle Datastream lit les événements lors de l'utilisation de la réplication basée sur GTID. |
| MySQL | primary_keys |
tableau de chaînes | tableau de chaînes | Liste des noms de colonnes (un ou plusieurs) qui constituent la clé primaire de la table. Si la table ne possède pas de clé primaire, ce champ reste vide. |
| MySQL | is_deleted |
boolean | boolean |
|
| MySQL | database |
chaîne | chaîne | Base de données associée à l'événement. |
| MySQL | table |
chaîne | chaîne | Table associée à l'événement. |
| MySQL | change_type |
chaîne | chaîne | Type de modification ( |
| Oracle | log_file |
chaîne | chaîne | Fichier journal à partir duquel Datastream extrait les événements dans la réplication CDC. |
| Oracle | scn |
long | long | Position du journal (décalage) dans le journal des transactions Oracle. |
| Oracle | row_id |
chaîne | string | row_id d'Oracle. |
| Oracle | is_deleted |
boolean | boolean |
|
| Oracle | database |
chaîne | chaîne | Base de données associée à l'événement. |
| Oracle | schema |
chaîne | chaîne | Schéma associé à la table de l'événement. |
| Oracle | table |
chaîne | chaîne | Table associée à l'événement. |
| Oracle | change_type |
chaîne | chaîne | Type de modification ( |
| Oracle | tx_id |
chaîne | chaîne | ID de la transaction à laquelle l'événement appartient. |
| Oracle | rs_id |
chaîne | chaîne | ID du jeu d'enregistrements. L'association de rs_id et ssn identifie de manière unique une ligne dans V$LOGMNR_CONTENTS. rs_id identifie de manière unique l'enregistrement de rétablissement qui a généré la ligne. |
| Oracle | ssn |
long | long | Numéro de séquence SQL. Ce numéro est utilisé avec rs_id et identifie de manière unique une ligne dans V$LOGMNR_CONTENTS. |
| PostgreSQL | schema |
string | chaîne | Schéma associé à la table de l'événement. |
| PostgreSQL | table |
string | chaîne | Table associée à l'événement. |
| PostgreSQL | is_deleted |
booléen | boolean |
|
| PostgreSQL | change_type |
string | string | Type de modification (INSERT, UPDATE, DELETE) représenté par l'événement.
|
| PostgreSQL | tx_id |
string | chaîne | ID de la transaction à laquelle l'événement appartient. |
| PostgreSQL | lsn |
string | string | Numéro de séquence du journal pour l'entrée actuelle. |
| PostgreSQL | primary_keys |
tableau de chaînes | tableau de chaînes | Liste des noms de colonnes (un ou plusieurs) qui constituent la clé primaire de la table. Si la table ne possède pas de clé primaire, ce champ reste vide. |
| SQL Server | table |
string | chaîne | Table associée à l'événement. |
| SQL Server | database |
long | long | Base de données associée à l'événement. |
| SQL Server | schema |
tableau de chaînes | tableau de chaînes | Schéma associé à la table de l'événement. |
| SQL Server | is_deleted |
booléen | boolean |
|
| SQL Server | lsn |
string | string | Numéro de séquence du journal pour l'événement. |
| SQL Server | tx_id |
string | chaîne | ID de la transaction à laquelle l'événement appartient. |
| SQL Server | physical_location |
tableau d'entiers | tableau d'entiers | Emplacement physique de l'enregistrement de journal décrit par trois entiers : l'ID de fichier, l'ID de page et l'ID d'emplacement de l'enregistrement. |
| SQL Server | replication_index |
tableau de chaînes | tableau de chaînes | Liste des noms de colonnes d'un index permettant d'identifier de manière unique une ligne de la table. |
| SQL Server | change_type |
string | string | Type de modification ( |
| Spanner | commit_timestamp |
timestamp-micros | string | Code temporel de commit de l'événement. |
| Spanner | snapshot |
booléen | booléen | Indique si l'événement est un événement d'instantané de remplissage. |
| Spanner | project_id |
string | string | Identifiant du projet Spanner. |
| Spanner | instance_id |
string | string | Identifiant de l'instance Spanner. |
| Spanner | database_id |
string | string | Identifiant de la base de données Spanner. |
| Spanner | schema |
string | chaîne | Schéma associé à la table de l'événement. Vide si la table ne se trouve pas dans un schéma nommé. |
| Spanner | table |
string | string | La table Spanner. |
| Spanner | change_stream_name |
string | string | Le flux de modifications Spanner. |
| Spanner | server_transaction_id |
string | string | Identifiant de transaction associé à l'événement. |
| Spanner | record_sequence |
long ou string | long ou string | Séquence d'enregistrement associée à l'événement. Pour les flux de modifications créés avec le mode de partition IMMUTABLE_KEY_RANGE, la valeur est un numéro de séquence numérique de type long. Pour les flux de modifications créés avec le mode de partition MUTABLE_KEY_RANGE, la valeur est la chaîne record_sequence de l'enregistrement de modification des données, par exemple 1cac8f3f38eeaf53-00000000.
|
| Spanner | mod_number |
long | long | Numéro de modification associé à l'événement. |
| Spanner | transaction_tag |
string | string | Tag de la transaction associée. |
| Spanner | number_of_records_in_transaction |
long | long | Nombre d'enregistrements dans la transaction associée. |
| Spanner | value_capture_type |
string | string | Type de capture de valeur du flux de modifications. Les valeurs possibles sont NEW_ROW et NEW_ROW_AND_OLD_VALUES. |
| Spanner | change_type |
string | string | Type de modification (INSERT, UPDATE ou DELETE) représenté par l'événement. |
| Spanner | primary_keys |
tableau de chaînes | tableau de chaînes | Liste des colonnes de clé primaire de la table. |
| Spanner | is_deleted |
booléen | booléen | Indique si la modification représente un événement DELETE. |
| Salesforce | object_name |
string | string | Nom de l'objet Salesforce associé à l'événement. |
| Salesforce | domain |
string | string | Nom du domaine associé à l'événement. |
| Salesforce | is_deleted |
booléen | boolean |
|
| Salesforce | change_type |
string | string | Type de modification ( |
| Salesforce | primary_keys |
tableau de chaînes | tableau de chaînes | Liste des noms de colonnes qui constituent la clé primaire de la table. Si la table ne possède pas de clé primaire, ce champ reste vide. |
| MongoDB | database |
string | chaîne | Base de données associée à l'événement. |
| MongoDB | collection |
string | string | Collection associée à l'événement. Les collections sont analogues aux tables dans les bases de données relationnelles. |
| MongoDB | change_type |
string | string | Type de modification (CREATE, UPDATE et DELETE) représenté par l'événement. |
| MongoDB | is_deleted |
booléen | boolean |
|
| MongoDB | primary_keys |
tableau de chaînes | tableau de chaînes | Le champ _id, qui sert de clé primaire pour chaque document d'une collection. |
Exemple de flux d'événements
Ce flux illustre les événements générés par trois opérations consécutives : INSERT, UPDATE et DELETE, sur une seule ligne d'une table SAMPLE pour une base de données source.
| TIME | THIS_IS_MY_PK (int) | FIELD1 (nchar pouvant être nul) | FIELD2 (nchar non-null)> |
|---|---|---|---|
| 0 | 1231535353 | foo | TLV |
| 1 | 1231535353 | NULL | TLV |
INSERT (T0)
La charge utile du message se compose de l'intégralité de la nouvelle ligne.
{
"stream_name": "projects/myProj/locations/myLoc/streams/Oracle-to-Source",
"read_method": "oracle-cdc-logminer",
"object": "SAMPLE.TBL",
"uuid": "d7989206-380f-0e81-8056-240501101100",
"read_timestamp": "2019-11-07T07:37:16.808Z",
"source_timestamp": "2019-11-07T02:15:39",
"source_metadata": {
"log_file": ""
"scn": 15869116216871,
"row_id": "AAAPwRAALAAMzMBABD",
"is_deleted": false,
"database": "DB1",
"schema": "ROOT",
"table": "SAMPLE"
"change_type": "INSERT",
"tx_id":
"rs_id": "0x0073c9.000a4e4c.01d0",
"ssn": 67,
},
"payload": {
"THIS_IS_MY_PK": "1231535353",
"FIELD1": "foo",
"FIELD2": "TLV",
}
}
UPDATE (T1)
La charge utile du message se compose de l'intégralité de la nouvelle ligne. Il n'inclut pas les valeurs précédentes.
{
"stream_name": "projects/myProj/locations/myLoc/streams/Oracle-to-Source",
"read_method": "oracle-cdc-logminer",
"object": "SAMPLE.TBL",
"uuid": "e6067366-1efc-0a10-a084-0d8701101101",
"read_timestamp": "2019-11-07T07:37:18.808Z",
"source_timestamp": "2019-11-07T02:17:39",
"source_metadata": {
"log_file":
"scn": 15869150473224,
"row_id": "AAAGYPAATAAPIC5AAB",
"is_deleted": false,
"database":
"schema": "ROOT",
"table": "SAMPLE"
"change_type": "UPDATE",
"tx_id":
"rs_id": "0x006cf4.00056b26.0010",
"ssn": 0,
},
"payload": {
"THIS_IS_MY_PK": "1231535353",
"FIELD1": null,
"FIELD2": "TLV",
}
}
DELETE (T2)
La charge utile du message se compose de l'intégralité de la nouvelle ligne.
{
"stream_name": "projects/myProj/locations/myLoc/streams/Oracle-to-Source",
"read_method": "oracle-cdc-logminer",
"object": "SAMPLE.TBL",
"uuid": "c504f4bc-0ffc-4a1a-84df-6aba382fa651",
"read_timestamp": "2019-11-07T07:37:20.808Z",
"source_timestamp": "2019-11-07T02:19:39",
"source_metadata": {
"log_file":
"scn": 158691504732555,
"row_id": "AAAGYPAATAAPIC5AAC",
"is_deleted": true,
"database":
"schema": "ROOT",
"table": "SAMPLE"
"change_type": "DELETE",
"tx_id":
"rs_id": "0x006cf4.00056b26.0011",
"ssn": 0,
},
"payload": {
"THIS_IS_MY_PK": "1231535353",
"FIELD1": null,
"FIELD2": "TLV",
}
}
Mise en ordre et cohérence
Cette section explique comment Datastream gère l'ordre et la cohérence.
commandes
Datastream ne garantit pas la livraison des événements dans l'ordre, mais chaque événement contient la ligne de données complète, l'horodatage de l'écriture des données dans la source et les métadonnées d'ordre.
Pour les événements qui font partie d'un backfill initial, le champ read_method est défini sur backfill ou se termine par -backfill, -fulldump ou -incremental. Les événements de remplissage sont générés de manière désordonnée par conception et peuvent être consommés dans n'importe quel ordre.
Pour les événements qui font partie de la réplication de capture de données modifiées (CDC) en cours, le champ read_method est défini sur cdc ou sur une valeur CDC spécifique à la source, telle que mysql-cdc-binlog, oracle-cdc-logminer, postgres-cdc-wal, sqlserver-cdc ou salesforce-cdc. La façon dont vous ordonnez les événements de réplication en cours sur toutes les sources dépend de votre destination :
| Destination | Description |
|---|---|
| BigQuery |
L'ordre des événements dans BigQuery dépend du mode d'écriture que vous configurez pour votre flux :
|
| Cloud Storage |
Dans Cloud Storage, les événements d'une même période peuvent s'étendre sur plusieurs fichiers. Pour déterminer l'ordre dans lequel les événements se sont produits au niveau de la source, triez-les à l'aide du champ de tableau |
Cohérence
Datastream s'assure que les données de la base de données source sont transmises à la destination au moins une fois. Aucun événement n'est manqué, mais il est possible que des événements soient dupliqués dans le flux. La fenêtre des événements en double doit être de l'ordre de quelques minutes, et l'identifiant unique universel (UUID) de l'événement dans les métadonnées d'événement peut être utilisé pour détecter les doublons.
Toutefois, ne vous fiez pas uniquement à l'UUID pour dédupliquer les événements en aval. En cas d'interruption du pipeline, Datastream relit le journal source à partir du dernier point de contrôle enregistré pour garantir la diffusion. Étant donné que Datastream génère des UUID de manière dynamique lorsqu'il lit un événement, ce type de nouvelle tentative génère de nouveaux UUID pour les mêmes données sources. De plus, lors d'opérations groupées dans la base de données source, plusieurs lignes distinctes peuvent partager le même numéro de séquence de modification (CSN).
Pour dédupliquer les données de manière fiable, partitionnez les événements en fonction des clés primaires de la table source, puis regroupez-les à l'aide du numéro de séquence de modification (CSN) (champ change_sequence_number). Cela permet de s'assurer que les nouvelles tentatives du même événement sont correctement réduites, tout en préservant les mises à jour de lignes légitimement distinctes des transactions groupées.
Lorsque les fichiers journaux de la base de données contiennent des transactions non validées, si des transactions sont annulées, la base de données le reflète dans les fichiers journaux en tant qu'opérations de langage de manipulation de données (LMD) "inverses". Par exemple, une opération INSERT annulée aura une opération DELETE correspondante. Datastream lit ces opérations à partir des fichiers journaux.
À propos des flux
Chaque flux comporte des métadonnées qui décrivent à la fois le flux et la source à partir de laquelle il extrait les données. Ces métadonnées incluent des informations telles que le nom du flux, ainsi que les profils de connexion source et de destination.
Pour afficher la définition complète de l'objet Stream, consultez la documentation de référence de l'API.
État d'un flux
Un flux peut avoir l'un des états suivants :
Not startedStartingRunningDrainingPausedFailedFailed permanently
Vous pouvez utiliser les journaux pour trouver des informations d'état supplémentaires, telles que le remplissage des tables ou le nombre de lignes traitées. Vous pouvez également utiliser l'API FetchStreamErrors pour récupérer les erreurs.
Métadonnées d'objets disponibles en utilisant l'API Discovery
L'API Discover renvoie des objets qui représentent la structure des objets définis dans la source de données ou la destination représentée par le profil de connexion. Chaque objet possède des métadonnées sur l'objet lui-même, ainsi que pour chaque champ de données qu'il extrait. Ces métadonnées sont disponibles via l'API Discover.
Étapes suivantes
- Pour en savoir plus sur les flux, consultez Cycle de vie des flux.
- Pour savoir comment créer un flux, consultez Créer un flux.