Gerenciar streams

Visão geral

Nesta seção, você aprenderá a usar a API Datastream para:

  • Criar streams
  • Receber informações sobre streams e objetos de stream
  • Atualizar streams iniciando, pausando, retomando e modificando-os, bem como iniciando e interrompendo o preenchimento de objetos de stream
  • Recuperar streams com falha permanente
  • Ativar streaming de objetos grandes para fluxos do Oracle
  • Excluir fluxos

Há duas maneiras de usar a API Datastream. É possível fazer chamadas de API REST ou usar a CLI (CLI) do Google Cloud.

Para informações de alto nível sobre como usar gcloud para gerenciar streams do Datastream, consulte gcloud Datastream streams.

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 lidar com seu caso de uso específico, use estes exemplos com a documentação de referência da API do Datastream.

Esta seção aborda os seguintes casos de uso:

Exemplo 1: fazer streaming de 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
  • Preencha todos os objetos incluídos no stream

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.

Como a solicitação não inclui o parâmetro customerManagedEncryptionKey, o sistema interno de gerenciamento de chaves do Google Cloud é usado para criptografar seus dados em vez de CMEK.

O parâmetro backfillAll associado à execução do preenchimento histórico (ou snapshot) está 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": {
      "dataFreshness": "900s",
      "sourceHierarchyDatasets": {
        "datasetTemplate": {
          "location": "us",
          "datasetIdPrefix": "prefix_"
        }
      }
    }
  },
  "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 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, você precisa definir mais dois campos 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 em que você quer replicar as mudanças. O nome da publicação precisa existir no banco de dados antes de iniciar um stream. A publicação precisa incluir, no mínimo, as tabelas especificadas na lista de includeObjects do 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: fazer streaming para um destino do Cloud Storage

Neste exemplo, você vai aprender a:

  • Streaming do Oracle para o Cloud Storage
  • Definir um conjunto de objetos para incluir no stream
  • Definir a 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.

Usando o parâmetro backfillNone, a solicitação especifica que apenas as alterações em andamento são transmitidas para o destino, sem preenchimento.

A solicitação especifica o parâmetro da 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 dados transmitidos da origem para o destino. Ele também especifica o keyring da sua CMEK.

Para mais informações sobre keyrings, consulte Recursos do Cloud KMS. Para mais informações sobre como proteger seus dados com chaves de criptografia, consulte o 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.

Validar a definição de um fluxo

Antes de criar um stream, é possível validar a definição dele. Dessa forma, é possível garantir que todas as verificações de validação sejam aprovadas e que o stream seja executado 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 fluxo, 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 fluxo, o Datastream verificará se a chave existe e se tem permissões para usá-la. Se uma dessas condições não for atendida, quando você validar o stream, a seguinte mensagem de erro será retornada:

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 acima, a verificação de CMEK_VALIDATE_PERMISSIONS não vai mais retornar uma mensagem de erro, mas vai 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 saber mais sobre como usar gcloud para recuperar informações sobre seu stream, clique aqui.

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 saber mais sobre como usar gcloud para recuperar informações sobre todos os seus streams, clique aqui.

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

Se quiser saber mais sobre como usar gcloud para recuperar informações sobre todos os objetos do stream, clique aqui.

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, clique aqui.

Iniciar um fluxo

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 um stream, altere 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

Se quiser mais informações sobre como usar gcloud para iniciar sua transmissão, clique aqui.

Iniciar um stream em uma posição específica

É possível iniciar um stream em uma posição específica para fontes MySQL e Oracle. Por exemplo, caso você queira realizar um preenchimento usando uma ferramenta externa, ou iniciar o CDC a partir de um arquivo de registro indicado. Para uma fonte do MySQL, é necessário indicar uma posição binlog. Para uma origem Oracle, um número de alteração do sistema (SCN, na sigla em inglês) no arquivo de redo log.

O código a seguir mostra uma solicitação para iniciar um stream já criado em uma posição específica.

Inicie um stream a partir de uma posição binlog específica (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 em que você quer iniciar seu 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 começará a ler a partir 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": 2
      }
    }
  }
}

gcloud

Não é possível iniciar uma transmissão em 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.

Inicie um stream com base em um número de alteração específico do sistema 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
      }
    }
  }
}
Substitua scn pelo número de alteração do sistema (SCN, na sigla em inglês) no arquivo de refazer registros a partir do qual 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": {
      "oracleScnPosition": {
        "scn": 123123
      }
    }
  }
}

gcloud

Não é possível iniciar uma transmissão em 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.

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

Se quiser mais informações sobre como usar gcloud para pausar a transmissão, clique aqui.

Retomar um fluxo

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 o gcloud para retomar sua transmissão, clique aqui.

Recuperar uma transmissão

É possível recuperar um stream com falha permanente para uma fonte MySQL, Oracle ou PostgreSQL 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 MySQL ou Oracle

Os exemplos de código a seguir mostram solicitações para recuperar um stream de uma origem MySQL ou Oracle de várias posições de arquivo de registros:

REST

Recupere 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": {}
  }
}

Recupere um stream de uma posição específica (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 do qual 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 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
      }
    }
  }
}

Recupere 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
      }
    }
  }
}
Substitua scn pelo número de alteração do sistema (SCN, na sigla em inglês) no arquivo de redo log do qual você quer recuperar seu stream. Este campo é obrigatório.

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 uma transmissão 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 fonte 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

Exemplo:

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run

Se você quiser alterar o slot de replicação, atualize o stream com o novo nome do slot:

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 uma transmissão usando gcloud.

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

Se quiser mais informações sobre como usar o gcloud para modificar seu stream, clique aqui.

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 o preenchimento inicial de um objeto do seu stream, clique aqui.

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 seu stream, clique aqui.

Alterar o número máximo de tarefas de CDC simultâneas

O código a seguir mostra como definir o número máximo de tarefas simultâneas de captura de dados alterados (CDC) para um stream do MySQL como 7.

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 de CDC simultâneas do valor anterior para 7. É possível usar valores de 0 a 50 (inclusive). Se você não definir o valor ou se defini-lo como 0, o padrão do sistema de cinco 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, clique aqui.

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 fluxo 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 de tarefas de preenchimento simultâneas máximas do valor anterior para 25. É possível usar valores de 0 a 50 (inclusive). Se você não definir o valor ou se defini-lo 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, clique aqui.

Ativar streaming de objetos grandes para origens Oracle

É possível ativar o streaming de objetos grandes, como objetos binários grandes (BLOB), objetos grandes de caracteres (CLOB) e objetos grandes de caracteres nacionais (NCLOB) para fluxos com fontes Oracle. A sinalização streamLargeObjects permite incluir objetos grandes em streams novos e atuais. A sinalização é definida no nível do fluxo. Não é preciso 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 fluxo

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

Se quiser mais informações sobre como usar o gcloud para excluir sua transmissão, clique aqui.

Anterior
Gerenciar perfis de conexão usando a API
Próxima
Resolver problemas com um stream