A API Datastream permite automatizar os fluxos de trabalho de replicação de dados por chamadas de API REST ou pela Google Cloud CLI. É possível criar, validar, modificar e excluir streams ou recuperar e listar streams e objetos de stream.
Também é possível iniciar, pausar,
retomar ou recuperar streams, gerenciar o preenchimento de objetos, ajustar a simultaneidade de tarefas e fazer streaming de objetos grandes do Oracle. Para detalhes de referência de comandos, consulte gcloud
datastream streams. Para um tutorial completo, consulte Usar APIs do Datastream.
Criar um stream
Nesta seção, você vai aprender a criar um stream usado para transferir dados da origem para um destino. Os exemplos a seguir não são abrangentes, mas destacam recursos específicos do Datastream. Para atender ao seu caso de uso específico, use esses exemplos com a documentação de referência da API do Datastream.
Esta seção aborda os seguintes casos de uso:
- Fazer streaming do Oracle para o Cloud Storage
- Fazer streaming do MySQL para o BigQuery
- Fazer streaming do PostgreSQL para o BigQuery
- Definir um conjunto de objetos para incluir no stream
- Preencher todos os objetos incluídos no stream
- Excluir objetos do stream
- Excluir objetos do preenchimento
- Definir CMEK para criptografar dados em repouso
- Definir o modo de gravação de um stream
- Fazer streaming para um projeto diferente no BigQuery
- Transmitir dados para tabelas do Apache Iceberg
- Fazer streaming do MongoDB usando o modo JSON estrito
Exemplo 1: transmitir objetos específicos para o BigQuery
Neste exemplo, você vai aprender a:
- Fazer streaming do MySQL para o BigQuery
- Incluir um conjunto de objetos no stream
- Definir o modo de gravação do stream como somente adição de dados
- Preencher todos os objetos incluídos no fluxo
Veja a seguir uma solicitação para extrair todas as tabelas de schema1 e duas tabelas específicas de schema2: tableA e tableC. Os eventos são gravados em um conjunto de dados
no BigQuery.
A solicitação não inclui o parâmetro customerManagedEncryptionKey. Portanto,
o sistema interno de gerenciamento de chaves do Google Cloud é usado para criptografar seus dados,
em vez da CMEK.
O parâmetro backfillAll associado à execução do preenchimento histórico (ou snapshot) é definido como um dicionário vazio ({}), o que significa que o Datastream preenche os dados históricos de todas as tabelas incluídas no fluxo.
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
Para mais informações sobre como usar gcloud para criar um stream, consulte a documentação do SDK Google Cloud.
Exemplo 2: excluir objetos específicos de um stream com uma origem do PostgreSQL
Neste exemplo, você vai aprender a:
- Fazer streaming do PostgreSQL para o BigQuery
- Excluir objetos do stream
- Excluir objetos do preenchimento
O código a seguir mostra uma solicitação para criar um stream usado para transferir dados de um banco de dados PostgreSQL de origem para o BigQuery. Ao criar um stream de um banco de dados PostgreSQL de origem, é necessário especificar dois campos adicionais específicos do PostgreSQL na solicitação:
replicationSlot: um slot de replicação é um pré-requisito para configurar um banco de dados PostgreSQL para replicação. É necessário criar um slot de replicação para cada stream.publication: uma publicação é um grupo de tabelas de que você quer replicar as alterações. O nome da publicação precisa existir no banco de dados antes de iniciar um stream. No mínimo, a publicação precisa incluir as tabelas especificadas na listaincludeObjectsdo stream.
O parâmetro backfillAll associado à execução do preenchimento histórico (ou snapshot) está definido para excluir uma tabela.
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
Para mais informações sobre como usar gcloud para criar um stream, consulte a documentação do SDK Google Cloud.
Exemplo 3: especificar o modo de gravação somente com adição de dados para um stream
Ao fazer streaming para o BigQuery, é possível definir o modo de gravação: merge ou
appendOnly. Para mais informações, consulte Configurar o modo de gravação.
Se você não especificar o modo de gravação na solicitação para criar um stream, o modo padrão
merge será usado.
A solicitação a seguir mostra como definir o modo appendOnly ao criar
um stream do MySQL para o 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
Para mais informações sobre como usar gcloud para criar um stream, consulte a documentação do SDK Google Cloud.
Exemplo 4: transmitir para um projeto diferente no BigQuery
Se você criou os recursos do Datastream em um projeto, mas quer fazer streaming para um projeto diferente no BigQuery, use uma solicitação semelhante à seguinte.
Se você especificar sourceHierarchyDatasets para o conjunto de dados de destino, preencha o campo projectId.
Se você especificar singleTargetDataset para o conjunto de dados de destino, preencha o campo datasetId no formato projectId:datasetId.
REST
Para obter 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": {} }
Para obter 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
Para obter 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
O conteúdo do arquivo de configuração source_hierarchy_cross_project_config.json:
{"sourceHierarchyDatasets": {"datasetTemplate": {"location": "us-central1", "datasetIdPrefix": "prefix_"}, "projectId": "myProjectId2"}}
Para obter 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
O conteúdo do arquivo de configuração single_target_cross_project_config.json:
{"singleTargetDataset": {"datasetId": "myProjectId2:myDatastetId"}}
Para mais informações sobre como usar gcloud para criar um stream, consulte a documentação do SDK Google Cloud.
Exemplo 5: transmitir para um destino do Cloud Storage
Neste exemplo, você vai aprender a:
- Transmitir do Oracle para o Cloud Storage
- Definir um conjunto de objetos para incluir no stream
- Definir CMEK para criptografar dados em repouso
A solicitação a seguir mostra como criar um stream que grava os eventos em um bucket no Cloud Storage.
Neste exemplo de solicitação, os eventos são gravados no formato de saída JSON, e um novo arquivo é criado a cada 100 MB ou 30 segundos (substituindo os valores padrão de 50 MB e 60 segundos).
Para o formato JSON, é possível:
Inclua um arquivo de esquema de tipos unificados no caminho. Como resultado, o Datastream grava dois arquivos no Cloud Storage: um arquivo de dados JSON e um arquivo de esquema Avro. O arquivo de esquema tem o mesmo nome do arquivo de dados, com uma extensão
.schema.Ative a compactação gzip para que o Datastream compacte os arquivos gravados no Cloud Storage.
Ao usar o parâmetro backfillNone, a solicitação especifica que apenas as mudanças em andamento são transmitidas para o destino, sem preenchimento.
A solicitação especifica o parâmetro de chave de criptografia gerenciada pelo cliente, que permite controlar as chaves usadas para criptografar dados em repouso em um projeto do Google Cloud . O parâmetro se refere à CMEK que o Datastream usa para criptografar os dados transmitidos da origem para o destino. Ele também especifica o keyring da CMEK.
Para mais informações sobre keyrings, consulte Recursos do Cloud KMS. Para mais informações sobre como proteger seus dados usando chaves de criptografia, consulte 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
Para mais informações sobre como usar gcloud para criar um stream, consulte a documentação do SDK Google Cloud.
Exemplo 6: fazer streaming para uma tabela do Apache Iceberg
Neste exemplo, você vai aprender a configurar um stream para replicar dados de um
banco de dados MySQL para uma tabela do Apache Iceberg no modo append-only.
Antes de criar a solicitação, verifique se você concluiu as seguintes
etapas:
- Ter um bucket do Cloud Storage em que você quer armazenar os dados.
- Criar uma conexão a recursos do Cloud
- Conceder à conexão a recursos do Cloud acesso ao bucket do Cloud Storage
Em seguida, use a seguinte solicitação para criar o 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
O conteúdo do arquivo de configuração de origem mysql_source_config.json:
{"excludeObjects": {}, "includeObjects": {"mysqlDatabases":[{ "database":"my-mysql-database"}]}}
O conteúdo do arquivo de configuração 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 {} }
Exemplo 7: fazer streaming do MongoDB usando o modo JSON estrito
Neste exemplo, você vai configurar um stream para replicar dados de uma origem do MongoDB para o BigQuery usando o modo JSON estrito.
Por padrão, o Datastream usa o modo JSON canônico (CANONICAL) para
priorizar a fidelidade dos dados. Se os aplicativos downstream exigirem compatibilidade
com analisadores JSON padrão, especifique o modo JSON estrito (STRICT).
Ao criar um stream de uma origem do MongoDB:
- Os valores válidos para
jsonModena API sãoCANONICAL(padrão) eSTRICT. - Na CLI gcloud, não há uma flag independente para o modo JSON.
Em vez disso, defina as configurações de origem do MongoDB em um arquivo de configuração
transmitido para a flag
--mongodb-source-config.
REST
Na API REST do Datastream, defina jsonMode como
"STRICT" no
objeto 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" } } } }
Substitua:
PROJECT_ID: o ID do seu projeto do Google Cloud .LOCATION: a região em que o stream é criado, comous-central1.STREAM_ID: o identificador do stream.STREAM_DISPLAY_NAME: um nome fácil de usar para o stream.SOURCE_PROFILE: o identificador do perfil de conexão de origem do MongoDB.DEST_PROFILE: o identificador do perfil de conexão de destino.DATABASE_NAME: o nome do banco de dados no cluster de origem do MongoDB.COLLECTION_NAME: o nome da coleção no banco de dados.DATASET_ID: o identificador do conjunto de dados de destino no BigQuery.
gcloud
Para criar um stream com uma origem do MongoDB e o modo JSON estrito usando a CLI gcloud:
- Crie um arquivo de configuração de origem no formato JSON ou YAML especificando
"jsonMode": "STRICT"e os objetos a serem replicados.
Arquivo de configuração JSON, por exemplo,
mongo_source_config.json:
{ "jsonMode": "STRICT", "includeObjects": { "databases": [ { "database": "DATABASE_NAME", "collections": [ { "collection": "COLLECTION_NAME" } ] } ] } }
Arquivo de configuração YAML, por exemplo,
mongo_source_config.yaml:
jsonMode: STRICT includeObjects: databases: - database: DATABASE_NAME collections: - collection: COLLECTION_NAME
- Transmita o arquivo de configuração para a
flag
--mongodb-source-configao criar o 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
O conteúdo do arquivo de configuração
bigquery_destination_config.json:
{ "singleTargetDataset": { "datasetId": "PROJECT_ID:DATASET_ID" } }
Substitua:
STREAM_ID: o identificador do stream.LOCATION: a região em que o stream é criado, comous-central1.STREAM_DISPLAY_NAME: um nome fácil de usar para o stream.PROJECT_ID: o ID do seu projeto do Google Cloud .SOURCE_PROFILE: o identificador do perfil de conexão de origem do MongoDB.DEST_PROFILE: o identificador do perfil de conexão de destino.DATABASE_NAME: o nome do banco de dados no cluster de origem do MongoDB.COLLECTION_NAME: o nome da coleção no banco de dados.DATASET_ID: o identificador do conjunto de dados de destino no BigQuery.
Para mais informações sobre como usar gcloud para criar um stream,
consulte a documentação do SDK Google Cloud.
Validar a definição de um stream
Antes de criar um stream, é possível validar a definição dele. Dessa forma, você garante que todas as verificações de validação sejam aprovadas e que o stream seja executado com sucesso quando criado.
Como validar verificações de stream:
- Indica se a fonte está configurada corretamente para permitir que o Datastream faça streaming de dados.
- Se a transmissão pode se conectar à origem e ao destino.
- A configuração completa do stream.
Para validar um stream, adicione &validate_only=true ao URL que precede o corpo da solicitação:
POST "https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/streams?streamId=STREAM_ID&validate_only=true"
Depois de fazer essa solicitação, você verá as verificações de validação que o Datastream executa para sua origem e seu destino e se as verificações serão aprovadas ou falharão. Para qualquer verificação de validação que não seja aprovada, serão exibidas informações sobre o motivo da falha e o que fazer para corrigir o problema.
Por exemplo, suponha que você tenha uma chave de criptografia gerenciada pelo cliente (CMEK) que quer que o Datastream use para criptografar dados transmitidos da origem para o destino. Como parte da validação do stream, o Datastream vai verificar se a chave existe e se tem permissões para usá-la. Se uma dessas condições não for atendida, a seguinte mensagem de erro será retornada quando você validar o stream:
CMEK_DOES_NOT_EXIST_OR_MISSING_PERMISSIONS
Para resolver esse problema, verifique se a chave fornecida existe e se a conta de serviço do Datastream tem a permissão cloudkms.cryptoKeys.get para a chave.
Depois de fazer as correções apropriadas, faça a solicitação novamente para garantir que todas as verificações de validação sejam aprovadas. No exemplo anterior, a verificação CMEK_VALIDATE_PERMISSIONS não vai mais retornar uma mensagem de erro, mas terá o status PASSED.
Receber informações sobre um fluxo
O código a seguir mostra uma solicitação para recuperar informações sobre um stream. Exemplos dessas informações:
- O nome do stream (identificador exclusivo)
- Um nome fácil de usar para o stream (nome de exibição)
- Carimbos de data/hora de quando o stream foi criado e atualizado pela última vez
- Informações sobre os perfis de conexão de origem e de destino associados ao stream
- O estado do stream
REST
GET https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/streams/STREAM_ID
A resposta é exibida da seguinte maneira:
{ "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
Para mais informações sobre como usar gcloud para recuperar informações sobre seu stream, consulte a documentação do SDK Google Cloud.
Listar fluxos
O código a seguir mostra uma solicitação para recuperar uma lista de todos os streams no projeto e no local especificados.
REST
GET https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams
gcloud
Para mais informações sobre como usar gcloud para recuperar informações sobre todas as suas transmissões, consulte a documentação do SDK Google Cloud.
Listar objetos de um fluxo
O código a seguir mostra uma solicitação para recuperar informações sobre todos os objetos de um stream.
REST
GET https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID/objects
gcloud
Para mais informações sobre como usar gcloud para recuperar informações sobre todos os objetos do stream, consulte a documentação do SDK Google Cloud.
A lista de objetos retornada pode ser semelhante a esta:
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
Para mais informações sobre como usar gcloud para listar objetos de um stream, consulte a documentação do SDK Google Cloud.
Iniciar um stream
O código a seguir mostra uma solicitação para iniciar um fluxo.
Usando o parâmetro updateMask na solicitação, apenas os campos que você especificar precisam ser incluídos no corpo da solicitação. Para iniciar uma transmissão, mude o valor no campo state de CREATED para RUNNING.
REST
PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID?updateMask=state { "state": "RUNNING" }
gcloud
Para mais informações sobre como usar gcloud para iniciar o stream, consulte a documentação do SDK Google Cloud.
Pausar um stream
O código a seguir mostra uma solicitação para pausar um fluxo em execução.
Neste exemplo, o campo especificado no parâmetro updateMask é state. Ao pausar o stream, você muda o estado dele de RUNNING para PAUSED.
REST
PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID?updateMask=state { "state": "PAUSED" }
gcloud
Para mais informações sobre como usar gcloud para pausar o stream, consulte a documentação do SDK Google Cloud.
Retomar um stream
O código a seguir mostra uma solicitação para retomar um stream pausado.
Neste exemplo, o campo especificado no parâmetro updateMask é state. Ao retomar o fluxo, você muda o estado dele de PAUSED de volta para RUNNING.
REST
PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID?updateMask=state { "state": "RUNNING" }
gcloud
Para mais informações sobre como usar gcloud para retomar o stream, consulte a documentação do SDK Google Cloud.
Recuperar um stream
É possível recuperar um stream com falha permanente usando o método RunStream. Cada
tipo de banco de dados de origem tem a própria definição de quais operações de recuperação de stream
são possíveis. Para mais informações, consulte Recuperar um stream.
Recuperar um stream para uma origem do MySQL ou Oracle
Os exemplos de código a seguir mostram solicitações para recuperar um stream de uma origem do MySQL ou da Oracle de várias posições de arquivo de registro:
REST
Recuperar um stream da posição atual. Esta é a opção padrão:
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID:run
Recuperar um stream da próxima posição disponível:
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID:run { "cdcStrategy": { "nextAvailableStartPosition": {} } }
Recuperar um stream da posição mais recente:
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID:run { "cdcStrategy": { "mostRecentStartPosition": {} } }
Recuperar um stream de uma posição específica (replicação baseada em binlog do 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 } } } }
Substitua:
- NAME_OF_THE_LOG_FILE: o nome do arquivo de registro de onde você quer recuperar o stream.
- POSITION: a posição no arquivo de registro em que você quer recuperar o stream. Se você não fornecer o valor, o Datastream vai recuperar o stream do cabeçalho do arquivo.
Exemplo:
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/ us-central1/streams/myStreamId1:run { "cdcStrategy": { "specificStartPosition": { "mysqlLogPosition": { "logFile": "binlog.001" "logPosition": 4 } } } }
Recuperar um stream de uma posição específica (replicação baseada em GTID do MySQL):
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID:run { "cdcStrategy": { "specificStartPosition": { "mysqlGtidPosition": { "gtidSet": "GTID_SET" } } } }
Substitua GTID_SET por um ou mais GTIDs únicos ou intervalos de GTIDs dos quais você quer recuperar o stream.
Exemplo:
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" } } } }
Recuperar um stream de uma posição específica (Oracle):
POST https://datastream.googleapis.com/v1/projects/[project-id]/locations/ [location]/streams/[stream-id]:run { "cdcStrategy": { "specificStartPosition": { "oracleScnPosition": { "scn": scn } } } }
Exemplo:
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
"cdcStrategy": {
"specificStartPosition": {
"oracleScnPosition": {
"scn": 234234
}
}
}
}
Para mais informações sobre as opções de recuperação disponíveis, consulte Recuperar um stream.
gcloud
Não é possível recuperar um stream usando gcloud.
Recuperar um stream para uma origem do PostgreSQL
O exemplo de código a seguir mostra uma solicitação para recuperar um stream de uma origem do PostgreSQL. Durante a recuperação, o stream começa a ler o primeiro número de sequência de registro (LSN, na sigla em inglês) no slot de replicação configurado para o stream.
REST
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID:run { "cdcStrategy": { "nextAvailableStartPosition": {} } }
Exemplo:
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/ us-central1/streams/myStreamId1:run
Se você quiser mudar o slot de replicação, atualize o stream com o novo nome do slot de replicação primeiro:
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
Não é possível recuperar um stream usando gcloud.
Recuperar um stream para uma origem do SQL Server
Os exemplos de código a seguir mostram solicitações de exemplo para recuperar um stream de uma origem do SQL Server.
REST
Recuperar um stream da primeira posição disponível:
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID:run
Exemplo:
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/ us-central1/streams/mySqlServerStreamId:run
Recuperar um stream de um número de sequência de registro preferido:
POST https://datastream.googleapis.com/v1/projects/[project-id]/locations/ [location]/streams/[stream-id]:run { "cdcStrategy": { "specificStartPosition": { "sqlServerLsnPosition": { "lsn": lsn } } } }
Exemplo:
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/mySqlServerStreamId:run
{
"cdcStrategy": {
"specificStartPosition": {
"sqlServerLsnPosition": {
"lsn": 0000123C:0000BA78:0004
}
}
}
}
gcloud
Não é possível recuperar um stream usando gcloud.
Iniciar ou retomar uma transmissão de uma posição específica
É possível iniciar ou retomar um stream pausado de uma posição específica para origens do MySQL, Oracle e SQL Server. Isso pode ser útil quando você quer realizar o preenchimento usando uma ferramenta externa ou iniciar a CDC em uma posição indicada:
- Para uma origem do MySQL, é necessário indicar uma posição de binlog ou um conjunto de GTID.
- Para uma origem do Oracle, é necessário indicar um número de alteração do sistema (SCN, na sigla em inglês) no arquivo de registro de redo log.
- Para uma origem do SQL Server, é necessário indicar um número de sequência de registro (LSN, na sigla em inglês) em uma tabela de alterações.
O código a seguir mostra exemplos de solicitações para iniciar ou retomar um stream já criado de uma posição específica.
Inicie ou retome um stream de uma posição específica do binlog (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 } } } }
Substitua:
- NAME_OF_THE_LOG_FILE: o nome do arquivo de registro de onde você quer iniciar o stream.
- POSITION: a posição no arquivo de registro em que você quer iniciar o stream. Se você não fornecer o valor, o Datastream vai começar a ler do início do arquivo.
Exemplo:
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/ us-central1/streams/myStreamId1:run { "cdcStrategy": { "specificStartPosition": { "mysqlLogPosition": { "logFile": "binlog.001" "logPosition": 2 } } } }
gcloud
Não é possível iniciar ou retomar um stream de uma posição específica usando gcloud. Para informações sobre como usar gcloud para iniciar ou retomar um stream, consulte a documentação do SDK Cloud.
Inicie ou retome uma transmissão de um conjunto de GTID específico (MySQL):
REST
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID:run { "cdcStrategy": { "specificStartPosition": { "mysqlGtidPosition": { "gtidSet": "GTID_SET" } } } }
Substitua GTID_SET por um ou mais GTIDs únicos ou intervalos de GTIDs em que você quer iniciar ou retomar o stream.
Exemplo:
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
Não é possível iniciar ou retomar um stream de uma posição específica usando gcloud. Para informações sobre como usar gcloud para iniciar ou retomar um stream, consulte a documentação do SDK Cloud.
Inicie ou retome um stream de um número de alteração do sistema específico no arquivo de redo log (Oracle):
REST
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID:run { "cdcStrategy": { "specificStartPosition": { "oracleScnPosition": { "scn": scn } } } }
Exemplo:
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
"cdcStrategy": {
"specificStartPosition": {
"oracleScnPosition": {
"scn": 123123
}
}
}
}
gcloud
Não é possível iniciar ou retomar um stream de uma posição específica usando gcloud. Para informações sobre como usar gcloud para iniciar uma transmissão, consulte a documentação do SDK Cloud.
Inicie ou retome um stream de um número de sequência de registro específico (SQL Server):
REST
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID:run { "cdcStrategy": { "specificStartPosition": { "sqlServerLsnPosition": { "lsn": "lsn" } } } }
Substitua lsn pelo número de sequência de registro na tabela de alterações em que você quer iniciar o stream. Este campo é obrigatório.
Exemplo:
POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
"cdcStrategy": {
"specificStartPosition": {
"sqlServerLsnPosition": {
"lsn": "0000123C0000BA780004"
}
}
}
}
Para mais informações, consulte a documentação de referência da API Datastream.
gcloud
Não é possível iniciar ou retomar um stream de uma posição específica usando gcloud. Para informações sobre como usar gcloud para iniciar um stream, consulte a documentação do SDK Cloud.
Modificar um stream
O código a seguir mostra uma solicitação para atualizar a configuração de rotação de arquivos de um fluxo para alterná-lo a cada 75 MB ou 45 segundos.
Neste exemplo, os campos especificados para o parâmetro updateMask incluem os campos fileRotationMb e fileRotationInterval, representados pelas sinalizações destinationConfig.gcsDestinationConfig.fileRotationMb e destinationConfig.gcsDestinationConfig.fileRotationInterval, respectivamente.
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 } } }
O código a seguir mostra uma solicitação para incluir um arquivo de esquema de tipos unificados no caminho dos arquivos que o Datastream grava no Cloud Storage. Como resultado, o Datastream grava dois arquivos: um arquivo de dados JSON e um arquivo de esquema Avro.
Neste exemplo, o campo especificado é jsonFileFormat, representado pela sinalização 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" } } } }
O código a seguir mostra uma solicitação para que o Datastream replique dados existentes, além de alterações contínuas nos dados, do banco de dados de origem para o destino.
A seção oracleExcludedObjects do código mostra as tabelas e esquemas que não podem ser preenchidos no destino.
Neste exemplo, todas as tabelas e esquemas serão preenchidos, exceto a tableA no 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
Para mais informações sobre como usar gcloud para modificar o stream, consulte a documentação do SDK Google Cloud.
Iniciar o preenchimento de um objeto de um fluxo
Um stream no Datastream pode preencher dados históricos e fazer streaming de alterações em andamento para um destino. As alterações em andamento serão sempre transmitidas de uma origem para um destino. No entanto, é possível especificar se você quer que os dados históricos sejam transmitidos.
Se você quiser que os dados históricos sejam transmitidos da origem para o destino, use o parâmetro backfillAll.
O Datastream também permite transmitir dados históricos apenas para tabelas de banco de dados específicas. Para fazer isso, use o parâmetro backfillAll e exclua as tabelas para as quais você não quer dados históricos.
Se você quiser que apenas as alterações em andamento sejam transmitidas no destino, use o parâmetro backfillNone. Se você quiser que o Datastream transmita um snapshot de todos os dados existentes da origem para o destino, inicie o preenchimento manualmente para os objetos que contêm esses dados.
Outro motivo para iniciar o preenchimento de um objeto é se os dados não estiverem sincronizados entre a origem e o destino. Por exemplo, um usuário pode excluir dados no destino acidentalmente, e os dados serão perdidos. Nesse caso, o preenchimento do objeto será iniciado como um "mecanismo de redefinição", porque todos os dados são transmitidos para o destino em uma única captura. Como resultado, os dados são sincronizados entre a origem e o destino.
Antes de iniciar o preenchimento de um objeto de um stream, você precisa recuperar informações sobre o objeto.
Cada objeto tem um OBJECT_ID, que identifica o objeto de maneira exclusiva. Use o OBJECT_ID para iniciar o preenchimento do stream.
REST
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID/objects/OBJECT_ID:startBackfillJob
gcloud
Para mais informações sobre como usar gcloud para iniciar o preenchimento de um objeto do stream, consulte a documentação do SDK Google Cloud.
Interromper o preenchimento de um objeto de um fluxo
Depois de iniciar o preenchimento de um objeto de um stream, você pode interromper o preenchimento do objeto. Por exemplo, se um usuário modificar um esquema de banco de dados, o esquema ou os dados poderão estar corrompidos. Você não quer que o esquema ou os dados sejam transmitidos para o destino, então o preenchimento do objeto é interrompido.
Também é possível interromper o preenchimento de um objeto para fins de balanceamento de carga. O Datastream pode executar vários preenchimentos em paralelo. Isso pode sobrecarregar a origem. Se a carga for significativa, interrompa o preenchimento de cada objeto e, em seguida, inicie o preenchimento para os objetos, um por um.
Antes de interromper o preenchimento de um objeto de um stream, é preciso fazer uma solicitação para recuperar informações sobre todos os objetos de um stream. Cada objeto retornado tem um OBJECT_ID, que identifica o objeto de maneira exclusiva. Use o OBJECT_ID para interromper o preenchimento do stream.
REST
POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID/objects/OBJECT_ID:stopBackfillJob
gcloud
Para mais informações sobre como usar gcloud para interromper o preenchimento de um objeto do stream, consulte a documentação do SDK Google Cloud.
Mudar o número máximo de tarefas simultâneas de CDC
O código a seguir mostra como definir como 7 o número máximo de tarefas simultâneas de captura de dados alterados (CDC, na sigla em inglês) para um stream do MySQL.
Neste exemplo, o campo especificado no parâmetro updateMask é maxConcurrentCdcTasks. Ao definir o valor como 7, você muda o número máximo de tarefas simultâneas de CDC do valor anterior para 7. É possível usar valores de 0 a 50 (inclusive). Se você não definir o valor ou se o definir como 0, o padrão do sistema de 5 tarefas será definido para o stream.
REST
PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID?updateMask=sourceConfig.mysqlSourceConfig.maxConcurrentCdcTasks { "sourceConfig": { "mysqlSourceConfig": { "maxConcurrentCdcTasks": "7" } } }
gcloud
Para mais informações sobre como usar gcloud, consulte a documentação do SDK Google Cloud.
Alterar o número máximo de tarefas de preenchimento simultâneas
O código a seguir mostra como definir o número máximo de tarefas de preenchimento simultâneas para um stream do MySQL como 25.
Neste exemplo, o campo especificado no parâmetro updateMask é maxConcurrentBackfillTasks. Ao definir o valor como 25, você muda o número máximo de tarefas de preenchimento simultâneas do valor anterior para 25. É possível usar valores de 0 a 50 (inclusive). Se você não definir o valor ou se o definir como 0, o padrão do sistema de 16 tarefas será definido para o stream.
REST
PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/ streams/STREAM_ID?updateMask=sourceConfig.mysqlSourceConfig.maxConcurrentBackfillTasks { "sourceConfig": { "mysqlSourceConfig": { "maxConcurrentBackfillTasks": "25" } } }
gcloud
Para mais informações sobre como usar gcloud, consulte a documentação do SDK Google Cloud.
Ativar o streaming de objetos grandes para origens do Oracle
É possível ativar o streaming de objetos grandes, como grandes objetos binários (BLOB),
grandes objetos de caracteres (CLOB) e grandes objetos de caracteres nacionais (NCLOB)
para streams com origens do Oracle. A flag streamLargeObjects permite incluir
objetos grandes em streams novos e atuais. A flag é definida no nível do stream,
não é necessário especificar as colunas de tipos de dados de objetos grandes.
O exemplo a seguir mostra como criar um stream que permite transmitir objetos grandes.
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
Para mais informações sobre como usar gcloud para atualizar um stream, consulte a documentação do SDK Google Cloud.
Excluir um stream
O código a seguir mostra uma solicitação para excluir um stream.
REST
DELETE https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/ LOCATION/streams/STREAM_ID
gcloud
Para mais informações sobre como usar gcloud para excluir o stream, consulte a documentação do SDK Google Cloud.
A seguir
- Saiba como usar a API Datastream para gerenciar perfis de conexão.
- Saiba como usar a API Datastream para gerenciar configurações de conectividade privada.
- Para mais informações sobre como usar a API Datastream, consulte a documentação de referência.