Criar e gerenciar streams de alterações

Nesta página, descrevemos como criar, modificar e visualizar fluxos de mudanças do Spanner para bancos de dados com dialeto GoogleSQL e PostgreSQL. Para saber mais sobre fluxo de alterações, consulte Sobre streams de alterações.

Como os fluxo de alterações são objetos de esquema, eles são criados e gerenciados pelas mesmas atualizações de esquema baseadas em DDL usadas para qualquer outro tipo de trabalho de definição de banco de dados, como criar tabelas ou adicionar índices.

O Spanner inicia uma operação de longa duração depois que você envia uma instrução DDL que muda o esquema, incluindo aquelas usadas para criar, alterar ou excluir fluxo de alterações. Um fluxo de mudanças novo ou alterado começa a monitorar as colunas ou tabelas especificadas pela nova configuração após a conclusão dessa operação de longa duração.

Criar um stream de alterações

Para criar um fluxo de alterações, é necessário informar o nome dele e os objetos de esquema que ele monitora: todo o banco de dados ou uma lista de tabelas e colunas específicas. Você também pode configurar um fluxo de mudanças com qualquer uma das seguintes opções:

• Especifique o período de retenção de dados para substituir o período padrão de um dia.
• Especifique o tipo de captura de valor para substituir o tipo de captura de valor padrão OLD_AND_NEW_VALUES.
• Aplique um filtro de exclusões com base em TTL para remover as exclusões com base em TTL dos seus fluxo de alterações.
• Aplique um filtro de modificações na tabela para excluir todas as modificações de tabela INSERT, UPDATE ou DELETE.

• Ative uma exclusão de registros no nível da transação para excluir determinadas transações dos seus fluxo de alterações.

CREATE CHANGE STREAM CHANGE_STREAM_NAME
  [FOR column_or_table_watching_definition[, ... ] ]
  [
    OPTIONS (
      retention_period = timespan,
      value_capture_type = type,
      exclude_ttl_deletes = boolean,
      exclude_insert = boolean,
      exclude_update = boolean,
      exclude_delete = boolean,
      allow_txn_exclusion = boolean
    )
  ]

CREATE CHANGE STREAM CHANGE_STREAM_NAME
  [FOR column_or_table_watching_definition[, ... ] ]
  [
    WITH (
      retention_period = timespan,
      value_capture_type = type,
      exclude_ttl_deletes = boolean,
      exclude_insert = boolean,
      exclude_update = boolean,
      exclude_delete = boolean,
      allow_txn_exclusion = boolean
    )
  ]

Um novo fluxo de alterações começa a monitorar os objetos de esquema atribuídos assim que a operação de longa duração que o criou é concluída.

Os exemplos a seguir ilustram a criação de fluxo de alterações com várias configurações.

Monitorar um banco de dados inteiro

Para criar um fluxo de alterações que monitora todas as mudanças de dados realizadas nas tabelas de um banco de dados, use a palavra-chave ALL:

CREATE CHANGE STREAM EverythingStream
FOR ALL;

A configuração ALL inclui implicitamente todas as tabelas e colunas de dados futuras do banco de dados assim que elas são criadas. Ele não inclui visualizações, tabelas de esquema de informações ou outros objetos além das tabelas de dados normais.

Monitorar tabelas específicas

Para limitar o escopo de um fluxo de alterações a tabelas específicas, em vez de um banco de dados inteiro, especifique uma lista de uma ou mais tabelas:

CREATE CHANGE STREAM SingerAlbumStream
FOR Singers, Albums;

O Spanner atualiza automaticamente os fluxos de mudanças que monitoram tabelas inteiras para refletir as mudanças de esquema que afetam essas tabelas, como colunas adicionadas ou removidas.

Acompanhar colunas específicas

Use a sintaxe table(column_1[, column_2, ...]) para acompanhar as mudanças em uma ou mais colunas específicas e não principais nas tabelas que você nomear:

CREATE CHANGE STREAM NamesAndTitles
FOR Singers(FirstName, LastName), Albums(Title);

Não é possível especificar colunas de chave primária aqui porque cada fluxo de mudanças sempre rastreia as chaves primárias de todas as tabelas que monitora. Isso permite que cada registro de mudança de dados identifique a linha alterada pela chave primária.

Assistir tabelas e colunas em um único fluxo

É possível combinar a sintaxe de observação de tabela e coluna dos dois exemplos anteriores em um único fluxo de mudanças:

CREATE CHANGE STREAM NamesAndAlbums
FOR Singers(FirstName, LastName), Albums;

Especificar um período de armazenamento mais longo

Para especificar um período de retenção de dados do fluxo de alterações maior que o padrão de um dia, defina retention_period como um período de até 30 dias, expresso como horas (h) ou dias (d).

Dois exemplos:

CREATE CHANGE STREAM LongerDataRetention
FOR ALL
OPTIONS ( retention_period = '36h' );

CREATE CHANGE STREAM MaximumDataRetention
FOR ALL
OPTIONS ( retention_period = '7d' );

CREATE CHANGE STREAM LongerDataRetention
FOR ALL
WITH ( retention_period = '36h' );

CREATE CHANGE STREAM MaximumDataRetention
FOR ALL
WITH ( retention_period = '7d' );

Especificar um tipo de captura de valor diferente

Para especificar um tipo de captura de valor de fluxo de mudanças diferente de OLD_AND_NEW_VALUES, defina value_capture_type como NEW_VALUES ou NEW_ROW, conforme mostrado nos exemplos a seguir:

CREATE CHANGE STREAM NewRowChangeStream
FOR ALL
OPTIONS ( value_capture_type = 'NEW_ROW' );

CREATE CHANGE STREAM NewValuesChangeStream
FOR ALL
OPTIONS ( value_capture_type = 'NEW_VALUES' );

CREATE CHANGE STREAM NewRowChangeStream
FOR ALL
WITH ( value_capture_type = 'NEW_ROW' );

CREATE CHANGE STREAM NewValuesChangeStream
FOR ALL
WITH ( value_capture_type = 'NEW_VALUES' );

Filtrar exclusões com base no TTL

É possível filtrar exclusões baseadas em TTL do escopo do fluxo de mudanças usando o filtro exclude_ttl_deletes.

Para mais informações sobre como esse filtro funciona, consulte Filtro de exclusões com base no tempo de vida.

CREATE CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
OPTIONS (exclude_ttl_deletes = true)

CREATE CHANGE STREAM NewFilterChangeStream FOR ALL
OPTIONS (exclude_ttl_deletes = true)

CREATE CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
WITH (exclude_ttl_deletes = true)

CREATE CHANGE STREAM NewFilterChangeStream FOR ALL
WITH (exclude_ttl_deletes = true)

Para adicionar ou remover o filtro de exclusões com base em TTL de um fluxo de mudanças, consulte Modificar o filtro de exclusões com base em TTL. Para confirmar seus filtros de fluxo de alterações, veja as definições do fluxo de alterações como DDL.

Filtrar por tipo de modificação da tabela

Filtre uma ou mais dessas modificações de tabela do escopo do fluxo de mudanças usando as seguintes opções de filtro disponíveis:

• exclude_insert: exclui todas as modificações da tabela INSERT
• exclude_update: exclui todas as modificações da tabela UPDATE
• exclude_delete: exclui todas as modificações da tabela DELETE

Para mais informações sobre como esses filtros funcionam, consulte Filtros de tipo de modificação de tabela.

CREATE CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
OPTIONS (MOD_TYPE_FILTER_NAME = true)

CREATE CHANGE STREAM NewFilterChangeStream FOR ALL
OPTIONS (exclude_insert = true, exclude_update = true)

CREATE CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
WITH (MOD_TYPE_FILTER_NAME = true)

CREATE CHANGE STREAM NewFilterChangeStream FOR ALL
WITH (exclude_insert = true, exclude_update = true)

Para adicionar ou remover um filtro de tipo de modificação de tabela de um fluxo de mudanças atual, consulte Modificar filtro por tipo de modificação de tabela. Para confirmar quais filtros de tipo de modificação de tabela existem para seu fluxo de alterações, veja as definições do fluxo de alterações como DDL.

Ativar a exclusão de registros no nível da transação

É possível ativar seus fluxos de alterações para excluir registros de transações de gravação especificadas definindo a opção allow_txn_exclusion ao criar um fluxo de alterações ou modificando um fluxo de alterações atual.

Para mais informações sobre como essa opção funciona, consulte Exclusão de registros no nível da transação.

CREATE CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
OPTIONS (allow_txn_exclusion = true)

CREATE CHANGE STREAM NewChangeStream FOR ALL
OPTIONS (allow_txn_exclusion = true)

CREATE CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
WITH (allow_txn_exclusion = true)

CREATE CHANGE STREAM NewChangeStream FOR ALL
WITH (allow_txn_exclusion = true)

Para ativar ou desativar a exclusão de registros no nível da transação de um fluxo de mudanças atual, consulte Modificar a exclusão de registros no nível da transação. Para verificar a configuração dessa opção, consulte ver definições de fluxo de alterações como DDL.

Especificar uma transação de gravação a ser excluída dos fluxo de alterações

Para especificar uma transação de gravação a ser excluída dos fluxo de alterações, defina o parâmetro exclude_txn_from_change_streams como true. Os exemplos de código a seguir mostram como especificar uma transação de gravação a ser excluída dos fluxo de alterações usando a biblioteca de cliente.

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/spanner"
)

// readWriteTxnExcludedFromChangeStreams executes the insert and update DMLs on
// Singers table excluded from tracking change streams with ddl option
// allow_txn_exclusion = true.
func readWriteTxnExcludedFromChangeStreams(w io.Writer, db string) error {
	// db = `projects/<project>/instances/<instance-id>/database/<database-id>`
	ctx := context.Background()
	client, err := spanner.NewClient(ctx, db)
	if err != nil {
		return fmt.Errorf("readWriteTxnExcludedFromChangeStreams.NewClient: %w", err)
	}
	defer client.Close()

	_, err = client.ReadWriteTransactionWithOptions(ctx, func(ctx context.Context, txn *spanner.ReadWriteTransaction) error {
		stmt := spanner.Statement{
			SQL: `INSERT Singers (SingerId, FirstName, LastName)
					VALUES (111, 'Virginia', 'Watson')`,
		}
		_, err := txn.Update(ctx, stmt)
		if err != nil {
			return fmt.Errorf("readWriteTxnExcludedFromChangeStreams.Update: %w", err)
		}
		fmt.Fprintln(w, "New singer inserted.")
		stmt = spanner.Statement{
			SQL: `UPDATE Singers SET FirstName = 'Hi' WHERE SingerId = 111`,
		}
		_, err = txn.Update(ctx, stmt)
		if err != nil {
			return fmt.Errorf("readWriteTxnExcludedFromChangeStreams.Update: %w", err)
		}
		fmt.Fprint(w, "Singer first name updated.")
		return nil
	}, spanner.TransactionOptions{ExcludeTxnFromChangeStreams: true})
	if err != nil {
		return err
	}
	return nil
}

static void readWriteTxnExcludedFromChangeStreams(DatabaseClient client) {
  // Exclude the transaction from allowed tracking change streams with alloww_txn_exclusion=true.
  // This exclusion will be applied to all the individual operations inside this transaction.
  client
      .readWriteTransaction(Options.excludeTxnFromChangeStreams())
      .run(
          transaction -> {
            transaction.executeUpdate(
                Statement.of(
                    "INSERT Singers (SingerId, FirstName, LastName)\n"
                        + "VALUES (1341, 'Virginia', 'Watson')"));
            System.out.println("New singer inserted.");

            transaction.executeUpdate(
                Statement.of("UPDATE Singers SET FirstName = 'Hi' WHERE SingerId = 111"));
            System.out.println("Singer first name updated.");

            return null;
          });
}

Modificar um fluxo de alterações

Para modificar a configuração de um fluxo de alterações, use uma instrução DDL ALTER CHANGE STREAM. Ele usa uma sintaxe semelhante a CREATE CHANGE STREAM. É possível mudar as colunas que um fluxo monitora ou a duração do período de armazenamento de dados dele. Você também pode suspender a observação por completo, mantendo os registros de mudança de dados.

Modificar o que um fluxo de alterações monitora

Este exemplo adiciona toda a tabela Songs ao fluxo de mudanças NamesAndAlbums configurado anteriormente:

ALTER CHANGE STREAM NamesAndAlbums
SET FOR Singers(FirstName, LastName), Albums, Songs;

O Spanner substitui o comportamento do fluxo de mudanças nomeado pela nova configuração após a conclusão da operação de longa duração que atualiza a definição do fluxo de mudanças no esquema do banco de dados.

Modificar o período de armazenamento de dados de um fluxo de alterações

Para modificar o período em que um fluxo de alterações retém os registros internos, defina o retention_period em uma instrução DDL ALTER CHANGE STREAM.

Este exemplo ajusta o período de armazenamento de dados para o fluxo de alterações NamesAndAlbums criado anteriormente:

ALTER CHANGE STREAM NamesAndAlbums
SET OPTIONS ( retention_period = '36h' );

ALTER CHANGE STREAM NamesAndAlbums
SET ( retention_period = '36h' );

Modificar o tipo de captura de valor de um fluxo de alterações

Para modificar o tipo de captura de valor de um fluxo de mudanças, defina a cláusula value_capture_type em uma instrução DDL ALTER CHANGE STREAM.

Este exemplo ajusta o tipo de captura de valor para NEW_VALUES.

ALTER CHANGE STREAM NamesAndAlbums
SET OPTIONS ( value_capture_type = 'NEW_VALUES' );

ALTER CHANGE STREAM NamesAndAlbums
SET ( value_capture_type = 'NEW_VALUES' );

Modificar o filtro de exclusões com base no TTL

Para modificar o filtro Exclusões com base em TTL de um fluxo de mudanças, defina o filtro exclude_ttl_deletes em uma instrução DDL ALTER CHANGE STREAM. Use isso para adicionar ou remover o filtro dos fluxo de alterações atuais.

Para mais informações sobre como esses filtros funcionam, consulte Filtro de exclusões com base no tempo de vida.

Adicionar o filtro de exclusões com base no TTL a um fluxo de mudanças atual

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET OPTIONS (exclude_ttl_deletes = true)

ALTER CHANGE STREAM NewFilterChangeStream FOR ALL
SET OPTIONS (exclude_ttl_deletes = true)

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET (exclude_ttl_deletes = true)

ALTER CHANGE STREAM NewFilterChangeStream FOR ALL
SET (exclude_ttl_deletes = true)

Remover o filtro de exclusões com base em TTL de um fluxo de alterações atual

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET OPTIONS (exclude_ttl_deletes = false)

ALTER CHANGE STREAM NewFilterChangeStream FOR ALL
SET OPTIONS (exclude_ttl_deletes = false)

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET (exclude_ttl_deletes = false)

ALTER CHANGE STREAM NewFilterChangeStream FOR ALL
SET (exclude_ttl_deletes = false)

Modificar filtro por tipo de modificação da tabela

Para modificar os filtros de tipo de modificação de tabela em um fluxo de alterações, defina o tipo de filtro em uma instrução DDL ALTER CHANGE STREAM. Você pode usar isso para adicionar ou remover um filtro do seu fluxo de mudanças.

Adicionar um filtro de tipo de modificação de tabela a um fluxo de alterações

ALTER CHANGE STREAM CHANGE_STREAM_NAME
SET OPTIONS (MOD_TYPE_FILTER_NAME = true)

ALTER CHANGE STREAM NewFilterChangeStream
SET OPTIONS (exclude_delete = true)

ALTER CHANGE STREAM CHANGE_STREAM_NAME
SET (MOD_TYPE_FILTER_NAME = true)

ALTER CHANGE STREAM NewFilterChangeStream
SET (exclude_delete = true)

Remover um filtro de tipo de modificação de tabela de um fluxo de alterações atual

ALTER CHANGE STREAM CHANGE_STREAM_NAME
SET OPTIONS (MOD_TYPE_FILTER_NAME = false)

ALTER CHANGE STREAM NewFilterChangeStream
SET OPTIONS (exclude_delete = false)

ALTER CHANGE STREAM CHANGE_STREAM_NAME
SET (MOD_TYPE_FILTER_NAME = false)

ALTER CHANGE STREAM NewFilterChangeStream
SET (exclude_delete = false)

Modificar o fluxo de alterações para permitir a exclusão de registros no nível da transação

É possível modificar o fluxo de alterações para que ele exclua registros de transações de gravação especificadas. Para fazer isso, defina a opção allow_txn_exclusion como true em uma instrução DDL ALTER CHANGE STREAM. Se você não definir essa opção ou se definir como false, o fluxo de mudanças vai monitorar todas as transações de gravação.

Para mais informações sobre como essa opção funciona, consulte Exclusão de registros no nível da transação.

Ativar a exclusão de registros no nível da transação para um fluxo de alterações atual

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET OPTIONS (allow_txn_exclusion = true)

ALTER CHANGE STREAM NewAllowedChangeStream FOR ALL
SET OPTIONS (allow_txn_exclusion = true)

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET (allow_txn_exclusion = true)

ALTER CHANGE STREAM NewAllowedChangeStream FOR ALL
SET (allow_txn_exclusion = true)

Desativar a exclusão de registros no nível da transação para um fluxo de alterações atual

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET OPTIONS (allow_txn_exclusion = false)

ALTER CHANGE STREAM NewFilterChangeStream FOR ALL
SET OPTIONS (allow_txn_exclusion = false)

ALTER CHANGE STREAM CHANGE_STREAM_NAME FOR ALL
SET (allow_txn_exclusion = false)

ALTER CHANGE STREAM NewAllowedChangeStream FOR ALL
SET (allow_txn_exclusion = false)

Suspender um fluxo de alterações

Se você quiser que um fluxo de mudanças interrompa a atividade, mas mantenha os registros internos (pelo menos durante o período de armazenamento de dados), altere para que ele não monitore nada.

Para fazer isso, emita uma instrução DDL ALTER CHANGE STREAM que substitua a definição do fluxo de mudanças pela frase especial DROP FOR ALL. Exemplo:

ALTER CHANGE STREAM MyStream DROP FOR ALL;

O fluxo continua existindo no banco de dados, mas não monitora objetos e não gera mais registros de mudança de dados. Os registros de mudança atuais permanecem intactos, sujeitos à política de retenção de dados do fluxo.

Para retomar um stream suspenso, emita outra instrução ALTER CHANGE STREAM com a configuração anterior.

Excluir um fluxo de alterações

Para excluir permanentemente um fluxo de alterações, emita uma instrução DROP CHANGE STREAM que inclua o nome do fluxo:

DROP CHANGE STREAM NamesAndAlbums;

O Spanner interrompe imediatamente o fluxo, remove-o do esquema do banco de dados e exclui os registros de alteração de dados.

Listar e visualizar fluxo de alterações

O console Google Cloud fornece uma interface da Web para listar e revisar as definições de fluxo de mudanças de um banco de dados. Também é possível conferir a estrutura dos fluxo de alterações como as instruções DDL equivalentes ou consultando o esquema de informações do banco de dados.

Ver fluxo de alterações com o console do Google Cloud

Para conferir uma lista dos fluxo de alterações de um banco de dados e analisar as definições deles:

1. Acesse a página de instâncias do Spanner no consoleGoogle Cloud .

   Abrir a página "Instâncias"

2. Navegue até a instância e o banco de dados adequados.

3. Clique em Fluxos de mudanças no menu de navegação.

Isso mostra uma lista de todos os fluxo de alterações do banco de dados e resume a configuração de cada um. Ao clicar no nome de um stream, mais detalhes sobre as tabelas e colunas monitoradas são exibidos.

Ver definições de fluxo de alterações como DDL

A visualização do esquema de um banco de dados como DDL inclui descrições de todos os fluxo de alterações, que aparecem como instruções CREATE CHANGE STREAM.

• Para fazer isso no console, clique no link Mostrar DDL equivalente na página do banco de dados no console do Google Cloud .

• Para fazer isso na linha de comando, use o comando ddl describe da Google Cloud CLI.

Consultar o esquema de informações sobre fluxo de alterações

É possível consultar diretamente o esquema de informações de um banco de dados sobre os fluxo de alterações. As tabelas a seguir contêm os metadados que definem os nomes dos fluxo de alterações, as tabelas e colunas que eles monitoram e os períodos de retenção:

• information_schema.change_streams

• information_schema.change_stream_tables

• information_schema.change_stream_columns

• information_schema.change_stream_options

Práticas recomendadas para change streams

Confira algumas práticas recomendadas para configurar e gerenciar fluxos de mudanças.

Considere um banco de dados de metadados separado

Os fluxos de alterações usam um banco de dados de metadados para manter o estado interno. O banco de dados de metadados pode ser igual ou diferente do banco de dados que contém os fluxo de alterações. Recomendamos criar um banco de dados separado para o armazenamento de metadados.

O conector fluxo de alterações do Spanner precisa de permissões de leitura e gravação no banco de dados de metadados. Não é necessário preparar esse banco de dados com um esquema. O conector faz isso.

Usar um banco de dados de metadados separado elimina complexidades que podem surgir ao permitir que o conector grave diretamente no banco de dados do aplicativo:

• Ao separar o banco de dados de metadados do banco de dados de produção com o fluxo de alterações, o conector só precisa de permissões de leitura para o banco de dados de produção.

• Ao restringir o tráfego do conector a um banco de dados de metadados separado, as gravações realizadas pelo próprio conector não são incluídas nos fluxo de alterações de produção. Isso é especialmente relevante para fluxo de alterações que monitoram todo o banco de dados.

Se nenhum banco de dados separado estiver sendo usado para armazenar os metadados, recomendamos monitorar o impacto na CPU do conector de fluxo de alterações nas instâncias.

Fazer o comparativo de novos fluxo de alterações e redimensionar se necessário

Antes de adicionar novos fluxo de alterações à sua instância de produção, faça um comparativo de uma carga de trabalho realista em uma instância de staging com os fluxos de mudanças ativados. Isso permite determinar se é necessário adicionar nós à instância para aumentar as capacidades de computação e armazenamento.

Execute esses testes até que as métricas de CPU e armazenamento se estabilizem. O ideal é que a utilização da CPU da instância permaneça abaixo dos valores máximos recomendados e que o uso de armazenamento não exceda o limite da instância.

Usar regiões diferentes para balancear a carga

Ao usar fluxo de alterações em uma configuração de instância multirregional, considere executar os pipelines de processamento em uma região diferente da região líder padrão. Isso ajuda a distribuir a carga de streaming entre réplicas não líderes. No entanto, se você precisar priorizar o menor atraso possível no streaming em vez do balanceamento de carga, execute a carga de streaming na região líder.

A seguir

• Criar pipelines de fluxo de alterações com o Dataflow.

• Use modelos para conectar rapidamente um fluxo de alterações ao BigQuery, Cloud Storage ou Pub/Sub.