L'API Datastream ti consente di automatizzare i flussi di lavoro di replica dei dati tramite chiamate API REST o Google Cloud CLI. Puoi creare, convalidare, modificare ed eliminare i flussi oppure recuperare ed elencare i flussi e gli oggetti dei flussi.
Puoi anche avviare, mettere in pausa, riprendere o recuperare i flussi, gestire il backfill degli oggetti, regolare la contemporaneità delle attività e eseguire lo streaming di oggetti Oracle di grandi dimensioni. Per i dettagli di riferimento dei comandi, consulta gcloud
datastream streams. Per una procedura dettagliata end-to-end, consulta Utilizza le API Datastream.
Crea uno stream
In questa sezione imparerai a creare uno stream utilizzato per trasferire i dati dall'origine a una destinazione. Gli esempi che seguono non sono esaustivi, ma evidenziano funzionalità specifiche di Datastream. Per affrontare il tuo caso d'uso specifico, utilizza questi esempi insieme alla documentazione di riferimento dell'API Datastream.
Questa sezione illustra i seguenti casi d'uso:
- Esegui lo streaming da Oracle a Cloud Storage
- Flusso di dati da MySQL a BigQuery
- Flusso di dati da PostgreSQL a BigQuery
- Definisci un insieme di oggetti da includere nello stream
- Esegui il backfill di tutti gli oggetti inclusi nello stream
- Escludi oggetti dallo stream
- Escludi oggetti dal backfill
- Definisci la CMEK per la crittografia dei dati at-rest
- Definisci la modalità di scrittura per uno stream
- Trasmetti in streaming a un progetto diverso in BigQuery
- Trasmetti i dati in streaming alle tabelle Apache Iceberg
- Trasmetti in streaming da MongoDB utilizzando la modalità JSON rigida
Esempio 1: trasmetti in streaming oggetti specifici a BigQuery
In questo esempio imparerai a:
- Flusso di dati da MySQL a BigQuery
- Includi un insieme di oggetti nello stream
- Definisci la modalità di scrittura per lo stream come di sola aggiunta
- Esegui il backfill di tutti gli oggetti inclusi nello stream
Di seguito è riportata una richiesta per estrarre tutte le tabelle da schema1 e due tabelle specifiche da schema2: tableA e tableC. Gli eventi vengono scritti in un set di dati in BigQuery.
La richiesta non include il parametro customerManagedEncryptionKey, pertanto per criptare i dati viene utilizzato il sistema di gestione delle chiavi interne di Google Cloud anziché CMEK.
Il parametro backfillAll associato all'esecuzione del backfill (o snapshot) storico è impostato su un dizionario vuoto ({}), il che significa che Datastream esegue il backfill dei dati storici di tutte le tabelle incluse nello stream.
REST
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams?streamId=mysqlCdcStream { "displayName": "MySQL CDC to BigQuery", "sourceConfig": { "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/streams/mysqlCp" , "mysqlSourceConfig": { "includeObjects": { "mysqlDatabases": [ { "database": "schema1" }, { "database": "schema2", "mysqlTables": [ { "table": "tableA", "table": "tableC" } ] } ] }, } }, "destinationConfig": { "destinationConnectionProfileName": "BigQueryCp", "bigqueryDestinationConfig": { "sourceHierarchyDatasets": { "datasetTemplate": { "location": "us", "datasetIdPrefix": "prefix_" } }, "dataFreshness": "900s" } }, "backfillAll": {} }
gcloud
Per ulteriori informazioni sull'utilizzo di gcloud per creare uno stream, consulta la documentazione di Google Cloud SDK.
Esempio 2: escludi oggetti specifici da un flusso con un'origine PostgreSQL
In questo esempio imparerai a:
- Flusso di dati da PostgreSQL a BigQuery
- Escludi oggetti dallo stream
- Escludi oggetti dal backfill
Il seguente codice mostra una richiesta per creare uno stream utilizzato per trasferire i dati da un database PostgreSQL di origine a BigQuery. Quando crei uno stream da un database PostgreSQL di origine, devi specificare due campi aggiuntivi specifici di PostgreSQL nella richiesta:
replicationSlot: uno slot di replica è un prerequisito per la configurazione di un database PostgreSQL per la replica. Devi creare uno slot di replica per ogni stream.publication: una pubblicazione è un gruppo di tabelle da cui vuoi replicare le modifiche. Il nome della pubblicazione deve esistere nel database prima di avviare uno stream. Come minimo, la pubblicazione deve includere le tabelle specificate nell'elencoincludeObjectsdel flusso.
Il parametro backfillAll associato all'esecuzione del backfill (o snapshot) storico è impostato in modo da escludere una tabella.
REST
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/ us-central1/streams?streamId=myPostgresStream { "displayName": "PostgreSQL to BigQueryCloud Storage", "sourceConfig": { "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/connectionProfiles/postgresCp", "postgresqlSourceConfig": { "replicationSlot": "replicationSlot1", "publication": "publicationA", "includeObjects": { "postgresqlSchemas": { "schema": "schema1" } }, "excludeObjects": { "postgresqlSchemas": [ { "schema": "schema1", "postgresqlTables": [ { "table": "tableA", "postgresqlColumns": [ { "column": "column5" } ] } ] } ] } } }, "destinationConfig": { "destinationConnectionProfileName": "BigQueryCp", "bigqueryDestinationConfig": { "dataFreshness": "900s", "sourceHierarchyDatasets": { "datasetTemplate": { "location": "us", "datasetIdPrefix": "prefix_" } } } }, "backfillAll": { "postgresqlExcludedObjects": { "postgresqlSchemas": [ { "schema": "schema1", "postgresqlTables": [ { "table": "tableA" } ] } ] } } }
gcloud
Per ulteriori informazioni sull'utilizzo di gcloud per creare uno stream, consulta la documentazione di Google Cloud SDK.
Esempio 3: specifica la modalità di scrittura di sola aggiunta per un flusso
Quando esegui lo streaming su BigQuery, puoi definire la modalità di scrittura: merge o
appendOnly. Per saperne di più, consulta Configura la modalità di scrittura.
Se non specifichi la modalità di scrittura nella richiesta di creazione di uno stream, viene utilizzata la modalità predefinita merge.
La seguente richiesta mostra come definire la modalità appendOnly quando crei un flusso da MySQL a BigQuery.
REST
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams?streamId=appendOnlyStream { "displayName": "My append-only stream", "sourceConfig": { "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/streams/mysqlCp" , "mysqlSourceConfig": { "includeObjects": { "mysqlDatabases": [ { "database": "myMySqlDb" } ] } } }, "destinationConfig": { "destinationConnectionProfileName": "BigQueryCp", "bigqueryDestinationConfig": { "sourceHierarchyDatasets": { "datasetTemplate": { "location": "us", "datasetIdPrefix": "prefix_" } }, "appendOnly": {} } }, "backfillAll": {} }
gcloud
Per ulteriori informazioni sull'utilizzo di gcloud per creare uno stream, consulta la documentazione di Google Cloud SDK.
Esempio 4: streaming in un progetto diverso in BigQuery
Se hai creato le risorse Datastream in un progetto, ma vuoi eseguire lo streaming in un progetto diverso in BigQuery, puoi farlo utilizzando una richiesta simile alla seguente.
Se specifichi sourceHierarchyDatasets per il set di dati di destinazione, devi compilare il campo projectId.
Se specifichi singleTargetDataset per il set di dati di destinazione, compila il campo datasetId nel formato projectId:datasetId.
REST
Per sourceHierarchyDatasets:
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams?streamId=crossProjectBqStream1 { "displayName": "My cross-project stream", "sourceConfig": { "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/streams/mysqlCp" , "mysqlSourceConfig": { "includeObjects": { "mysqlDatabases": [ { "database": "myMySqlDb" } ] } } }, "destinationConfig": { "destinationConnectionProfileName": "BigQueryCp", "bigqueryDestinationConfig": { "sourceHierarchyDatasets": { "datasetTemplate": { "location": "us", "datasetIdPrefix": "prefix_" }, "projectId": "myProjectId2" } } }, "backfillAll": {} }
Per singleTargetDataset:
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams?streamId=crossProjectBqStream2 { "displayName": "My cross-project stream", "sourceConfig": { "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/streams/mysqlCp" , "mysqlSourceConfig": { "includeObjects": { "mysqlDatabases": [ { "database": "myMySqlDb" } ] } } }, "destinationConfig": { "destinationConnectionProfileName": "BigQueryCp", "bigqueryDestinationConfig": { "singleTargetDataset": { "datasetId": "myProjectId2:myDatasetId" }, } }, "backfillAll": {} }
gcloud
Per sourceHierarchyDatasets:
datastream streams create crossProjectBqStream1 --location=us-central1 --display-name=my-cross-project-stream --source=source-cp --mysql-source-config=mysql_source_config.json --destination=destination-cp --bigquery-destination-config=source_hierarchy_cross_project_config.json --backfill-none
I contenuti del file di configurazione source_hierarchy_cross_project_config.json:
{"sourceHierarchyDatasets": {"datasetTemplate": {"location": "us-central1", "datasetIdPrefix": "prefix_"}, "projectId": "myProjectId2"}}
Per singleTargetDataset:
datastream streams create crossProjectBqStream --location=us-central1 --display-name=my-cross-project-stream --source=source-cp --mysql-source-config=mysql_source_config.json --destination=destination-cp --bigquery-destination-config=single_target_cross_project_config.json --backfill-none
I contenuti del file di configurazione single_target_cross_project_config.json:
{"singleTargetDataset": {"datasetId": "myProjectId2:myDatastetId"}}
Per ulteriori informazioni sull'utilizzo di gcloud per creare uno stream, consulta la documentazione di Google Cloud SDK.
Esempio 5: flussi di dati verso una destinazione Cloud Storage
In questo esempio imparerai a:
- Esegui lo streaming da Oracle a Cloud Storage
- Definisci un insieme di oggetti da includere nello stream
- Definisci la CMEK per la crittografia dei dati at-rest
La seguente richiesta mostra come creare uno stream che scrive gli eventi in un bucket in Cloud Storage.
In questa richiesta di esempio, gli eventi vengono scritti nel formato di output JSON e viene creato un nuovo file ogni 100 MB o 30 secondi (eseguendo l'override dei valori predefiniti di 50 MB e 60 secondi).
Per il formato JSON, puoi:
Includi un file di schema tipi unificati nel percorso. Di conseguenza, Datastream scrive due file in Cloud Storage: un file di dati JSON e un file di schema Avro. Il file dello schema ha lo stesso nome del file di dati, con l'estensione
.schema.Attiva la compressione gzip per consentire a Datastream di comprimere i file scritti in Cloud Storage.
Utilizzando il parametro backfillNone, la richiesta specifica che solo le modifiche in corso vengono trasmesse in streaming alla destinazione, senza backfill.
La richiesta specifica il parametro della chiave di crittografia gestita dal cliente, che ti consente di controllare le chiavi utilizzate per criptare i dati at-rest all'interno di un progetto Google Cloud . Il parametro si riferisce alla CMEK utilizzata da Datastream per criptare i dati trasmessi in streaming dall'origine alla destinazione. Specifica anche il keyring per la chiave CMEK.
Per saperne di più sulle chiavi automatizzate, consulta Risorse Cloud KMS. Per ulteriori informazioni sulla protezione dei dati utilizzando le chiavi di crittografia, consulta Cloud Key Management Service (KMS).
REST
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/ us-central1/streams?streamId=myOracleCdcStream { "displayName": "Oracle CDC to Cloud Storage", "sourceConfig": { "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/ connectionProfiles/OracleCp", "oracleSourceConfig": { "includeObjects": { "oracleSchemas": [ { "schema": "schema1" } ] } } }, "destinationConfig": { "destinationConnectionProfileName": "GcsBucketCp", "gcsDestinationConfig": { "path": "/folder1", "jsonFileFormat": { "schemaFileFormat": "AVRO_SCHEMA_FILE" }, "fileRotationMb": 100, "fileRotationInterval": 30 } }, "customerManagedEncryptionKey": "projects/myProjectId1/locations/us-central1/ keyRings/myRing/cryptoKeys/myEncryptionKey", "backfillNone": {} }
gcloud
Per ulteriori informazioni sull'utilizzo di gcloud per creare uno stream, consulta la documentazione di Google Cloud SDK.
Esempio 6: flussi di dati in una tabella Apache Iceberg
In questo esempio, imparerai a configurare uno stream per replicare i dati da un database MySQL a una tabella Apache Iceberg in modalità append-only.
Prima di creare la richiesta, assicurati di aver completato i seguenti
passaggi:
- Avere un bucket Cloud Storage in cui vuoi archiviare i dati
- Crea una connessione alle risorse Cloud
- Concedi l'accesso alla connessione alle risorse Cloud al bucket Cloud Storage
Puoi quindi utilizzare la seguente richiesta per creare lo stream:
REST
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams?streamId=mysqlIcebergStream { "displayName": "MySQL to Apache Iceberg stream", "sourceConfig": { "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/streams/mysqlIcebergCp" , "mysqlSourceConfig": { "includeObjects": { "mysqlDatabases": [ { "database": "my-mysql-database" } ] } } }, "destinationConfig": { "destinationConnectionProfileName": "projects/myProjectId1/locations/us-central1/connectionProfiles/my-bq-cp-id" , "bigqueryDestinationConfig": { "blmtConfig": { "bucket": "my-gcs-bucket-name", "rootPath": "my/folder", "connectionName": "my-project-id.us-central1.my-bigquery-connection-name", "fileFormat": "PARQUET", "tableFormat": "ICEBERG" }, "singleTargetDataset": { "datasetId": "my-project-id:my-bigquery-dataset-id" }, "appendOnly": {} } }, "backfillAll": {} }
gcloud
datastream streams create mysqlIcebergStream --location=us-central1 --display-name=mysql-to-bl-stream --source=source--mysql-source-config=mysql_source_config.json --destination=destination --bigquery-destination-config=bl_config.json --backfill-none
I contenuti del file di configurazione di origine mysql_source_config.json:
{"excludeObjects": {}, "includeObjects": {"mysqlDatabases":[{ "database":"my-mysql-database"}]}}
I contenuti del file di configurazione bl_config.json:
{ "blmtConfig": { "bucket": "my-gcs-bucket-name", "rootPath": "my/folder","connectionName": "my-project-id.us-central1.my-bigquery-connection-name", "fileFormat": "PARQUET", "tableFormat": "ICEBERG" }, "singleTargetDataset": {"datasetId": "my-project-id:my-bigquery-dataset-id"}, "appendOnly": {} }
Terraform
resource "google_datastream_stream" "stream" { stream_id = "mysqlBlStream" location = "us-central1" display_name = "MySQL to Apache Iceberg stream" source_config { source_connection_profile = "/projects/myProjectId1/locations/us-central1/streams/mysqlBlCp" mysql_source_config { include_objects { mysql_databases { database = "my-mysql-database" } } } } destination_config { destination_connection_profile ="projects/myProjectId1/locations/us-central1/connectionProfiles/my-bq-cp-id" bigquery_destination_config { single_target_dataset { dataset_id = "my-project-id:my-bigquery-dataset-id" } blmt_config { bucket = "my-gcs-bucket-name" table_format = "ICEBERG" file_format = "PARQUET" connection_name = "my-project-id.us-central1.my-bigquery-connection-name" root_path = "my/folder" } append_only {} } } backfill_none {} }
Esempio 7: flussi di dati da MongoDB utilizzando la modalità JSON rigorosa
In questo esempio, configuri uno stream per replicare i dati da un'origine MongoDB a BigQuery utilizzando la modalità JSON rigida.
Per impostazione predefinita, Datastream utilizza la modalità JSON canonica (CANONICAL) per dare priorità alla fedeltà dei dati. Se le tue applicazioni downstream richiedono la compatibilità con i parser JSON standard, puoi specificare la modalità JSON rigorosa (STRICT).
Quando crei uno stream da un'origine MongoDB:
- I valori validi per
jsonModenell'API sonoCANONICAL(predefinito) eSTRICT. - In gcloud CLI non esiste un flag autonomo per la modalità JSON.
Configura invece le impostazioni dell'origine MongoDB tramite un file di configurazione passato al flag
--mongodb-source-config.
REST
Nell'API REST Datastream, imposta jsonMode su
"STRICT" all'interno dell'oggetto
sourceConfig.mongodbSourceConfig:
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/streams?streamId=STREAM_ID { "displayName": "STREAM_DISPLAY_NAME", "sourceConfig": { "sourceConnectionProfile": "projects/PROJECT_ID/locations/LOCATION/connectionProfiles/SOURCE_PROFILE", "mongodbSourceConfig": { "jsonMode": "STRICT", "includeObjects": { "databases": [ { "database": "DATABASE_NAME", "collections": [ { "collection": "COLLECTION_NAME" } ] } ] } } }, "destinationConfig": { "destinationConnectionProfile": "projects/PROJECT_ID/locations/LOCATION/connectionProfiles/DEST_PROFILE", "bigqueryDestinationConfig": { "singleTargetDataset": { "datasetId": "PROJECT_ID:DATASET_ID" } } } }
Sostituisci quanto segue:
PROJECT_ID: l'ID del progetto Google Cloud .LOCATION: la regione in cui viene creato lo stream, ad esempious-central1.STREAM_ID: l'identificatore dello stream.STREAM_DISPLAY_NAME: un nome facile da usare per lo stream.SOURCE_PROFILE: l'identificatore del profilo di connessione MongoDB di origine.DEST_PROFILE: l'identificatore del profilo di connessione di destinazione.DATABASE_NAME: il nome del database nel cluster MongoDB di origine.COLLECTION_NAME: il nome della raccolta nel database.DATASET_ID: l'identificatore del set di dati di destinazione in BigQuery.
gcloud
Per creare uno stream con un'origine MongoDB e la modalità JSON rigida utilizzando gcloud CLI:
- Crea un file di configurazione di origine in formato JSON o YAML specificando
"jsonMode": "STRICT"e gli oggetti da replicare.
File di configurazione JSON, ad esempio mongo_source_config.json:
{ "jsonMode": "STRICT", "includeObjects": { "databases": [ { "database": "DATABASE_NAME", "collections": [ { "collection": "COLLECTION_NAME" } ] } ] } }
File di configurazione YAML, ad esempio mongo_source_config.yaml:
jsonMode: STRICT includeObjects: databases: - database: DATABASE_NAME collections: - collection: COLLECTION_NAME
- Passa il file di configurazione al flag
--mongodb-source-configdurante la creazione dello stream:
gcloud datastream streams create STREAM_ID \
--location=LOCATION \
--display-name="STREAM_DISPLAY_NAME" \
--source=projects/PROJECT_ID/locations/LOCATION/connectionProfiles/SOURCE_PROFILE \
--destination=projects/PROJECT_ID/locations/LOCATION/connectionProfiles/DEST_PROFILE \
--mongodb-source-config=mongo_source_config.json \
--bigquery-destination-config=bigquery_destination_config.json
I contenuti del file di configurazione bigquery_destination_config.json:
{ "singleTargetDataset": { "datasetId": "PROJECT_ID:DATASET_ID" } }
Sostituisci quanto segue:
STREAM_ID: l'identificatore dello stream.LOCATION: la regione in cui viene creato lo stream, ad esempious-central1.STREAM_DISPLAY_NAME: un nome facile da usare per lo stream.PROJECT_ID: l'ID del progetto Google Cloud .SOURCE_PROFILE: l'identificatore del profilo di connessione MongoDB di origine.DEST_PROFILE: l'identificatore del profilo di connessione di destinazione.DATABASE_NAME: il nome del database nel cluster MongoDB di origine.COLLECTION_NAME: il nome della raccolta nel database.DATASET_ID: l'identificatore del set di dati di destinazione in BigQuery.
Per ulteriori informazioni sull'utilizzo di gcloud per creare uno stream, consulta la documentazione di Google Cloud SDK.
Convalida la definizione di uno stream
Prima di creare un flusso, puoi convalidarne la definizione. In questo modo, puoi assicurarti che tutti i controlli di convalida vengano superati e che il flusso venga eseguito correttamente al momento della creazione.
La convalida di uno stream controlla:
- Se l'origine è configurata correttamente per consentire a Datastream di trasmettere i dati.
- Se lo stream può connettersi sia all'origine che alla destinazione.
- La configurazione end-to-end dello stream.
Per convalidare uno stream, aggiungi &validate_only=true all'URL che precede il corpo della richiesta:
POST "https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/streams?streamId=STREAM_ID&validate_only=true"
Dopo aver effettuato questa richiesta, vedrai i controlli di convalida eseguiti da Datastream per l'origine e la destinazione, nonché l'esito positivo o negativo dei controlli. Per ogni controllo di convalida non superato, vengono visualizzate informazioni sul motivo dell'errore e su cosa fare per risolvere il problema.
Ad esempio, supponiamo che tu abbia una chiave di crittografia gestita dal cliente (CMEK) che vuoi che Datastream utilizzi per criptare i dati trasmessi in streaming dall'origine alla destinazione. Nell'ambito della convalida dello stream, Datastream verificherà che la chiave esista e che Datastream disponga delle autorizzazioni per utilizzarla. Se una di queste condizioni non viene soddisfatta, durante la convalida dello stream verrà restituito il seguente messaggio di errore:
CMEK_DOES_NOT_EXIST_OR_MISSING_PERMISSIONS
Per risolvere il problema, verifica che la chiave fornita esista e che il account di servizio Datastream disponga dell'autorizzazione cloudkms.cryptoKeys.get per la chiave.
Dopo aver apportato le correzioni appropriate, effettua nuovamente la richiesta per assicurarti che tutti i controlli di convalida vengano superati. Per l'esempio precedente, il controllo CMEK_VALIDATE_PERMISSIONS non restituirà più un messaggio di errore, ma avrà lo stato PASSED.
Ottieni informazioni su uno stream
Il codice seguente mostra una richiesta per recuperare informazioni su uno stream. Queste informazioni comprendono:
- Il nome dello stream (identificatore univoco)
- Un nome facile da usare per lo stream (nome visualizzato)
- Timestamp della creazione e dell'ultimo aggiornamento dello stream
- Informazioni sui profili di connessione di origine e destinazione associati allo stream
- Lo stato dello stream
REST
GET https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/streams/STREAM_ID
La risposta è la seguente:
{ "name": "myOracleCdcStream", "displayName": "Oracle CDC to Cloud Storage", "createTime": "2019-12-15T15:01:23.045123456Z", "updateTime": "2019-12-15T15:01:23.045123456Z", "sourceConfig": { "sourceConnectionProfileName": "myOracleDb", "oracleSourceConfig": { "includeObjects": { "oracleSchemas": [ { "schema": "schema1" }, { "schema": "schema3", "oracleTables": [ { "table": "tableA" }, { "table": "tableC" } ] } ] } } }, "destinationConfig": { "destinationConnectionProfileName": "myGcsBucket", "gcsDestinationConfig": { "path": "/folder1", "avroFileFormat": {}, "fileRotationMb": 100, "fileRotationInterval": 60 } }, "state": "RUNNING" "customerManagedEncryptionKey": "projects/myProjectId1/locations/us-central1/ keyRings/myRing/cryptoKeys/myEncryptionKey", "backfillAll": {} }
gcloud
Per saperne di più sull'utilizzo di gcloud per recuperare informazioni sul tuo stream, consulta la documentazione di Google Cloud SDK.
Elenca stream
Il codice seguente mostra una richiesta per recuperare un elenco di tutti gli stream nel progetto e nella località specificati.
REST
GET https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams
gcloud
Per ulteriori informazioni sull'utilizzo di gcloud per recuperare informazioni su tutti i tuoi stream, consulta la documentazione di Google Cloud SDK.
Elenca gli oggetti di uno stream
Il codice seguente mostra una richiesta per recuperare informazioni su tutti gli oggetti di uno stream.
REST
GET https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID/objects
gcloud
Per ulteriori informazioni sull'utilizzo di gcloud per recuperare informazioni su tutti gli oggetti del tuo stream, consulta la documentazione di Google Cloud SDK.
L'elenco di oggetti restituito potrebbe essere simile al seguente:
REST
{ "streamObjects": [ { "name": "projects/myProjectId1/locations/us-central1/streams/myStream/ objects/object1", "displayName": "employees.salaries", "backfillJob": { "state": "ACTIVE", "trigger": "AUTOMATIC", "lastStartTime": "2021-10-18T12:12:26.344878Z" }, "sourceObject": { "mysqlIdentifier": { "database": "employees", "table": "salaries" } } }, { "name": "projects/myProjectId1/locations/us-central1/streams/myStream/ objects/object2", "displayName": "contractors.hours", "sourceObject": { "mysqlIdentifier": { "database": "contractors", "table": "hours" } } }, { "name": "projects/myProjectId1/locations/us-central1/streams/myStream/ objects/object3", "displayName": "employees.departments", "backfillJob": { "state": "COMPLETED", "trigger": "AUTOMATIC", "lastStartTime": "2021-10-18T11:26:12.869880Z", "lastEndTime": "2021-10-18T11:26:28.405653Z" }, "sourceObject": { "mysqlIdentifier": { "database": "employees", "table": "departments" } } } ] }
gcloud
Per ulteriori informazioni sull'utilizzo di gcloud per elencare gli oggetti di uno stream, consulta la documentazione di Google Cloud SDK.
Avvio di un flusso
Il seguente codice mostra una richiesta di avvio di uno stream.
Se utilizzi il parametro updateMask nella richiesta, nel corpo della richiesta devono essere inclusi solo i campi che specifichi. Per avviare uno stream, modifica il valore nel campo state da CREATED a RUNNING.
REST
PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID?updateMask=state { "state": "RUNNING" }
gcloud
Per saperne di più sull'utilizzo di gcloud per avviare lo stream, consulta la documentazione di Google Cloud SDK.
Pausa di un flusso
Il seguente codice mostra una richiesta di messa in pausa di uno stream in esecuzione.
In questo esempio, il campo specificato per il parametro updateMask è il campo state. Se metti in pausa lo stream, il suo stato cambia da RUNNING a PAUSED.
REST
PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID?updateMask=state { "state": "PAUSED" }
gcloud
Per saperne di più sull'utilizzo di gcloud per mettere in pausa lo stream, consulta la documentazione di Google Cloud SDK.
Ripresa di un flusso
Il seguente codice mostra una richiesta di ripresa di uno stream in pausa.
In questo esempio, il campo specificato per il parametro updateMask è il campo state. Se riprendi lo stream, il suo stato cambia da PAUSED a RUNNING.
REST
PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID?updateMask=state { "state": "RUNNING" }
gcloud
Per ulteriori informazioni sull'utilizzo di gcloud per riprendere lo stream, consulta la documentazione di Google Cloud SDK.
Recupera un flusso
Puoi recuperare uno stream con errore permanente utilizzando il metodo RunStream. Ogni tipo di database di origine ha una propria definizione delle operazioni di recupero dello stream possibili. Per saperne di più, consulta Recupera uno stream.
Recupera un flusso per un'origine MySQL o Oracle
I seguenti esempi di codice mostrano le richieste per recuperare un flusso per un'origine MySQL o Oracle da varie posizioni dei file di log:
REST
Recupera uno stream dalla posizione corrente. Questa è l'opzione predefinita:
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID:run
Recupera un flusso dalla successiva posizione disponibile:
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID:run { "cdcStrategy": { "nextAvailableStartPosition": {} } }
Recupera un flusso dalla posizione più recente:
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID:run { "cdcStrategy": { "mostRecentStartPosition": {} } }
Recupera un flusso da una posizione specifica (replica basata su binlog MySQL):
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID:run { "cdcStrategy": { "specificStartPosition": { "mysqlLogPosition": { "logFile": "NAME_OF_THE_LOG_FILE" "logPosition": POSITION } } } }
Sostituisci quanto segue:
- NAME_OF_THE_LOG_FILE: il nome del file di log da cui vuoi recuperare lo stream
- POSITION: la posizione nel file di log da cui vuoi recuperare lo stream. Se non fornisci il valore, Datastream recupera lo stream dall'inizio del file.
Ad esempio:
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/ us-central1/streams/myStreamId1:run { "cdcStrategy": { "specificStartPosition": { "mysqlLogPosition": { "logFile": "binlog.001" "logPosition": 4 } } } }
Recupera un flusso da una posizione specifica (replica basata su GTID MySQL):
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID:run { "cdcStrategy": { "specificStartPosition": { "mysqlGtidPosition": { "gtidSet": "GTID_SET" } } } }
Sostituisci GTID_SET con uno o più GTID singoli o intervalli di GTID da cui vuoi recuperare lo stream.
Ad esempio:
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/ us-central1/streams/myStreamId1:run { "cdcStrategy": { "specificStartPosition": { "mysqlGtidPosition": { "gtidSet": "22cc56f5-3862-379a-9af5-051c59baef9d:1-561143685:561143688-591036613,b8d7df02-832b-32b9-bec7-2018806b76f6:1-3" } } } }
Recupera un flusso da una posizione specifica (Oracle):
POST https://datastream.googleapis.com/v1/projects/[project-id]/locations/ [location]/streams/[stream-id]:run { "cdcStrategy": { "specificStartPosition": { "oracleScnPosition": { "scn": scn } } } }
Ad esempio:
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
"cdcStrategy": {
"specificStartPosition": {
"oracleScnPosition": {
"scn": 234234
}
}
}
}
Per saperne di più sulle opzioni di recupero disponibili, consulta Recupera un flusso.
gcloud
Il recupero di uno stream utilizzando gcloud non è supportato.
Recupera un flusso per un'origine PostgreSQL
Il seguente esempio di codice mostra una richiesta di recupero di uno stream per un'origine PostgreSQL. Durante il recupero, lo stream inizia a leggere dal primo numero di sequenza di log (LSN) nello slot di replica configurato per lo stream.
REST
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID:run { "cdcStrategy": { "nextAvailableStartPosition": {} } }
Ad esempio:
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/ us-central1/streams/myStreamId1:run
Se vuoi modificare lo slot di replica, aggiorna prima lo stream con il nuovo nome dello slot di replica:
PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID?updateMask=sourceConfig.postgresqlSourceConfig.replicationSlot { "sourceConfig": { "postgresqlSourceConfig": { "replicationSlot": "NEW_REPLICATION_SLOT_NAME" } } }
gcloud
Il recupero di uno stream utilizzando gcloud non è supportato.
Recupera un flusso per un'origine SQL Server
I seguenti esempi di codice mostrano richieste di esempio per recuperare uno stream per un'origine SQL Server.
REST
Recupera uno stream dalla prima posizione disponibile:
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID:run
Ad esempio:
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/ us-central1/streams/mySqlServerStreamId:run
Recupera un flusso da un numero di sequenza di log preferito:
POST https://datastream.googleapis.com/v1/projects/[project-id]/locations/ [location]/streams/[stream-id]:run { "cdcStrategy": { "specificStartPosition": { "sqlServerLsnPosition": { "lsn": lsn } } } }
Ad esempio:
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/mySqlServerStreamId:run
{
"cdcStrategy": {
"specificStartPosition": {
"sqlServerLsnPosition": {
"lsn": 0000123C:0000BA78:0004
}
}
}
}
gcloud
Il recupero di uno stream utilizzando gcloud non è supportato.
Avvia o riprendi uno stream da una posizione specifica
Puoi avviare un flusso o riprendere un flusso in pausa da una posizione specifica per le origini MySQL, Oracle e SQL Server. Questo può essere utile quando vuoi eseguire il backfill utilizzando uno strumento esterno o avviare CDC da una posizione che indichi:
- Per un'origine MySQL, devi indicare una posizione binlog o un set GTID.
- Per un'origine Oracle, devi indicare un numero della modifica di sistema (SCN) nel file redo log.
- Per un'origine SQL Server, devi indicare un numero di sequenza di log (LSN) in una tabella delle modifiche.
Il seguente codice mostra esempi di richieste per avviare o riprendere uno stream già creato da una posizione specifica.
Avvia o riprendi uno stream da una posizione binlog specifica (MySQL):
REST
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID:run { "cdcStrategy": { "specificStartPosition": { "mysqlLogPosition": { "logFile": "NAME_OF_THE_LOG_FILE" "logPosition": POSITION } } } }
Sostituisci quanto segue:
- NAME_OF_THE_LOG_FILE: il nome del file di log da cui vuoi avviare lo stream.
- POSITION: la posizione nel file di log da cui vuoi avviare lo stream. Se non fornisci il valore, Datastream inizia a leggere dall'inizio del file.
Ad esempio:
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/ us-central1/streams/myStreamId1:run { "cdcStrategy": { "specificStartPosition": { "mysqlLogPosition": { "logFile": "binlog.001" "logPosition": 2 } } } }
gcloud
L'avvio o la ripresa di uno stream da una posizione specifica utilizzando gcloud non è supportato. Per informazioni sull'utilizzo di gcloud per avviare o riprendere uno stream, consulta la documentazione di Cloud SDK.
Avvia o riprendi uno stream da un set GTID specifico (MySQL):
REST
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID:run { "cdcStrategy": { "specificStartPosition": { "mysqlGtidPosition": { "gtidSet": "GTID_SET" } } } }
Sostituisci GTID_SET con uno o più GTID singoli o intervalli di GTID da cui vuoi avviare o riprendere lo stream.
Ad esempio:
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/ us-central1/streams/myStreamId1:run { "cdcStrategy": { "specificStartPosition": { "mysqlGtidPosition": { "gtidSet": "22cc56f5-3862-379a-9af5-051c59baef9d:1-561143685:561143688-591036613,b8d7df02-832b-32b9-bec7-2018806b76f6:3-7" } } } }
gcloud
L'avvio o la ripresa di uno stream da una posizione specifica utilizzando gcloud non è supportato. Per informazioni sull'utilizzo di gcloud per avviare o riprendere uno stream, consulta la documentazione di Cloud SDK.
Avvia o riprendi un flusso da un numero della modifica di sistema specifico nel file di redo log (Oracle):
REST
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID:run { "cdcStrategy": { "specificStartPosition": { "oracleScnPosition": { "scn": scn } } } }
Ad esempio:
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
"cdcStrategy": {
"specificStartPosition": {
"oracleScnPosition": {
"scn": 123123
}
}
}
}
gcloud
L'avvio o la ripresa di uno stream da una posizione specifica utilizzando gcloud non è supportato. Per informazioni sull'utilizzo di gcloud per avviare uno stream, consulta la documentazione di Cloud SDK.
Avvia o riprendi uno stream da un numero di sequenza di log specifico (SQL Server):
REST
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID:run { "cdcStrategy": { "specificStartPosition": { "sqlServerLsnPosition": { "lsn": "lsn" } } } }
Sostituisci lsn con il numero di sequenza del log nella tabella delle modifiche da cui vuoi avviare lo stream. Questo campo è obbligatorio.
Ad esempio:
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
"cdcStrategy": {
"specificStartPosition": {
"sqlServerLsnPosition": {
"lsn": "0000123C0000BA780004"
}
}
}
}
Per saperne di più, consulta la documentazione di riferimento dell'API Datastream.
gcloud
L'avvio o la ripresa di uno stream da una posizione specifica utilizzando gcloud non è supportato. Per informazioni sull'utilizzo di gcloud per avviare uno stream, consulta la documentazione di Cloud SDK.
Modifica di un flusso
Il codice seguente mostra una richiesta di aggiornamento della configurazione di rotazione dei file di uno stream per ruotare il file ogni 75 MB o 45 secondi.
In questo esempio, i campi specificati per il parametro updateMask includono i campi fileRotationMb e fileRotationInterval, rappresentati rispettivamente dai flag destinationConfig.gcsDestinationConfig.fileRotationMb e destinationConfig.gcsDestinationConfig.fileRotationInterval.
REST
PATCH https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams/myStream?updateMask=destinationConfig.gcsDestinationConfig. fileRotationMb,destinationConfig.gcsDestinationConfig.fileRotationInterval { "destinationConfig": { "gcsDestinationConfig": { "fileRotationMb": 75, "fileRotationInterval": 45 } } }
Il seguente codice mostra una richiesta per includere un file di schema tipi unificati nel percorso dei file che Datastream scrive in Cloud Storage. Di conseguenza, Datastream scrive due file: un file di dati JSON e un file di schema Avro.
In questo esempio, il campo specificato è jsonFileFormat, rappresentato dal flag destinationConfig.gcsDestinationConfig.jsonFileFormat.
PATCH https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams/myStream?updateMask=destinationConfig.gcsDestinationConfig. jsonFileFormat { "destinationConfig": { "gcsDestinationConfig": { "jsonFileFormat" { "schemaFileFormat": "AVRO_SCHEMA_FILE" } } } }
Il codice seguente mostra una richiesta a Datastream di replicare i dati esistenti, oltre alle modifiche in corso ai dati, dal database di origine alla destinazione.
La sezione oracleExcludedObjects del codice mostra le tabelle e gli schemi per cui è vietato il backfill nella destinazione.
In questo esempio, verrà eseguito il backfill di tutte le tabelle e di tutti gli schemi, ad eccezione di tableA in schema3.
PATCH https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams/myStream?updateMask=backfillAll { "backfillAll": { "oracleExcludedObjects": { "oracleSchemas": [ { "schema": "schema3", "oracleTables": [ { "table": "tableA" } ] } ] } } }
gcloud
Per ulteriori informazioni sull'utilizzo di gcloud per modificare lo stream, consulta la documentazione di Google Cloud SDK.
Avvia il backfill per un oggetto di uno stream
Un flusso in Datastream può eseguire il backfill dei dati storici, nonché trasmettere in streaming le modifiche in corso a una destinazione. Le modifiche in corso verranno sempre trasmesse in streaming da un'origine a una destinazione. Tuttavia, puoi specificare se vuoi che i dati storici vengano trasmessi in streaming.
Se vuoi che i dati storici vengano trasmessi dall'origine alla destinazione, utilizza il parametro backfillAll.
Datastream consente anche di trasmettere in streaming i dati storici solo per tabelle di database specifiche. Per farlo, utilizza il parametro backfillAll ed escludi le tabelle per le quali non vuoi dati storici.
Se vuoi che solo le modifiche in corso vengano trasmesse in streaming alla destinazione, utilizza il parametro backfillNone. Se poi vuoi che Datastream trasmetta uno snapshot di tutti i dati esistenti dall'origine alla destinazione, devi avviare manualmente il backfill per gli oggetti che contengono questi dati.
Un altro motivo per avviare il backfill per un oggetto è la mancata sincronizzazione dei dati tra l'origine e la destinazione. Ad esempio, un utente può eliminare inavvertitamente i dati nella destinazione, che vengono persi. In questo caso, l'avvio del backfill per l'oggetto funge da "meccanismo di ripristino" perché tutti i dati vengono trasmessi in streaming alla destinazione in un'unica soluzione. Di conseguenza, i dati vengono sincronizzati tra l'origine e la destinazione.
Prima di poter avviare il backfill per un oggetto di uno stream, devi recuperare le informazioni sull'oggetto.
Ogni oggetto ha un OBJECT_ID, che lo identifica in modo univoco. Utilizza OBJECT_ID per avviare il backfill per lo stream.
REST
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID/objects/OBJECT_ID:startBackfillJob
gcloud
Per ulteriori informazioni sull'utilizzo di gcloud per avviare il backfill per un oggetto del tuo stream, consulta la documentazione di Google Cloud SDK.
Interrompi il backfill per un oggetto di uno stream
Dopo aver avviato il backfill per un oggetto di uno stream, puoi interromperlo. Ad esempio, se un utente modifica uno schema del database, lo schema o i dati potrebbero essere danneggiati. Non vuoi che questo schema o questi dati vengano trasmessi alla destinazione, quindi interrompi il backfill per l'oggetto.
Puoi anche interrompere il backfill di un oggetto per scopi di bilanciamento del carico. Datastream può eseguire più backfill in parallelo. Ciò potrebbe comportare un carico aggiuntivo sull'origine. Se il carico è significativo, interrompi il backfill per ogni oggetto e poi avvialo per gli oggetti, uno alla volta.
Prima di poter interrompere il backfill per un oggetto di uno stream, devi effettuare una richiesta per recuperare informazioni su tutti gli oggetti di uno stream. Ogni oggetto restituito ha un OBJECT_ID, che lo identifica in modo univoco. Utilizza OBJECT_ID per interrompere il backfill per lo stream.
REST
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID/objects/OBJECT_ID:stopBackfillJob
gcloud
Per ulteriori informazioni sull'utilizzo di gcloud per interrompere il backfill di un oggetto del tuo stream, consulta la documentazione di Google Cloud SDK.
Modifica il numero massimo di attività CDC contemporanee
Il seguente codice mostra come impostare su 7 il numero massimo di attività di Change Data Capture (CDC) simultanee per uno stream MySQL.
In questo esempio, il campo specificato per il parametro updateMask è il campo maxConcurrentCdcTasks. Impostando il valore su 7, modifichi il numero di attività CDC contemporanee massime dal valore precedente a 7. Puoi utilizzare valori compresi tra 0 e 50 (inclusi). Se non definisci il valore o se lo definisci come 0, per lo stream viene impostato il valore predefinito di sistema di 5 attività.
REST
PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID?updateMask=sourceConfig.mysqlSourceConfig.maxConcurrentCdcTasks { "sourceConfig": { "mysqlSourceConfig": { "maxConcurrentCdcTasks": "7" } } }
gcloud
Per saperne di più sull'utilizzo di gcloud, consulta la documentazione di Google Cloud SDK.
Modifica il numero massimo di attività di backfill simultanee
Il seguente codice mostra come impostare su 25 il numero massimo di attività di backfill simultanee per uno stream MySQL.
In questo esempio, il campo specificato per il parametro updateMask è il campo maxConcurrentBackfillTasks. Impostando il valore su 25, modifichi il numero di attività di backfill contemporanee massime dal valore precedente a 25. Puoi utilizzare valori compresi tra 0 e 50 (inclusi). Se non definisci il valore o se lo definisci come 0, per lo stream viene impostato il valore predefinito di sistema di 16 attività.
REST
PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/ streams/STREAM_ID?updateMask=sourceConfig.mysqlSourceConfig.maxConcurrentBackfillTasks { "sourceConfig": { "mysqlSourceConfig": { "maxConcurrentBackfillTasks": "25" } } }
gcloud
Per saperne di più sull'utilizzo di gcloud, consulta la documentazione di Google Cloud SDK.
Abilita lo streaming di oggetti di grandi dimensioni per le origini Oracle
Puoi attivare lo streaming di oggetti di grandi dimensioni, come oggetti binari di grandi dimensioni (BLOB), oggetti di grandi dimensioni di caratteri (CLOB) e oggetti di grandi dimensioni di caratteri nazionali (NCLOB) per gli stream con origini Oracle. Il flag streamLargeObjects consente di includere oggetti di grandi dimensioni sia negli stream nuovi che in quelli esistenti. Il flag è impostato a livello di stream,
non è necessario specificare le colonne dei tipi di dati di oggetti di grandi dimensioni.
L'esempio seguente mostra come creare uno stream che consente di trasmettere oggetti di grandi dimensioni.
REST
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/ us-central1/streams?streamId=myOracleLobStream { "displayName": "Oracle LOB stream", "sourceConfig": { "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/connectionProfiles/OracleCp" , "oracleSourceConfig": { "includeObjects": { "oracleSchemas": [ { "schema": "schema1", "oracleTables": [ { "table": "tableA", "oracleColumns": [ { "column": "column1,column2" } ] } ] } ] }, "excludeObjects": {}, "streamLargeObjects": {} } } }
gcloud
Per saperne di più sull'utilizzo di gcloud per aggiornare uno stream, consulta la documentazione di Google Cloud SDK.
Eliminazione di un flusso
Il codice seguente mostra una richiesta di eliminazione di uno stream.
REST
DELETE https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID
gcloud
Per ulteriori informazioni sull'utilizzo di gcloud per eliminare lo stream, consulta la documentazione di Google Cloud SDK.
Passaggi successivi
- Scopri come utilizzare l'API Datastream per gestire i profili di connessione.
- Scopri come utilizzare l'API Datastream per gestire le configurazioni di connettività privata.
- Per ulteriori informazioni sull'utilizzo dell'API Datastream, consulta la documentazione di riferimento.