Introdução a fluxos¶

Armazenamento offset¶

Quando criado, um fluxo tira logicamente um instantâneo inicial de cada linha do objeto de origem (por exemplo, tabela, tabela externa ou as tabelas subjacentes para uma exibição) inicializando um ponto no tempo (chamado offset) como a versão transacional atual do objeto. O sistema de rastreamento de alterações utilizado pelo fluxo então registra informações sobre as alterações do DML após esse instantâneo ter sido tirado. Os registros de alterações fornecem o estado de uma linha antes e depois da alteração. As informações de alteração espelham a estrutura da coluna do objeto de origem rastreado e incluem colunas adicionais de metadados que descrevem cada evento de alteração.

Note que um fluxo em si não contém quaisquer dados de tabela. Um fluxo só armazena um offset para o objeto de origem e retorna registros CDC aproveitando o histórico de versões para o objeto de origem. Quando o primeiro fluxo para uma tabela é criado, várias colunas ocultas são adicionadas à tabela de origem e começam a armazenar metadados de rastreamento de alterações. Estas colunas consomem uma pequena quantidade de armazenamento. Os registros CDC retornados ao consultar um fluxo dependem de uma combinação do offset armazenado no fluxo e dos metadados de rastreamento de alterações armazenados na tabela. Observe que para Streams on Views, o rastreamento de alterações deve ser habilitado explicitamente para a exibição e tabelas subjacentes para adicionar as colunas ocultas a estas tabelas.

Pode ser útil pensar em um fluxo como um marcador, que indica um ponto no tempo nas páginas de um livro (ou seja, o objeto de origem). Um marcador pode ser descartado e outros marcadores podem ser inseridos em lugares diferentes em um livro. Da mesma forma, um fluxo pode ser descartado e outros fluxos podem ser criados no mesmo momento ou em momentos diferentes (seja criando os fluxos consecutivamente em momentos diferentes ou usando o Time Travel) para consumir os registros de alterações de um objeto no mesmo offset ou em offsets diferentes.

Um exemplo de um consumidor de registros CDC é um pipeline de dados, no qual apenas os dados em tabelas de preparação que mudaram desde a última extração são transformados e copiados para outras tabelas.

Versões de tabela¶

Uma nova versão da tabela é criada sempre que uma transação que inclui uma ou mais instruções DML é confirmada na tabela. Isto se aplica aos seguintes tipos de tabela:

• Tabelas padrão

• Tabelas de diretório

• Tabelas externas

• Tabelas subjacentes para uma exibição

No histórico de transações de uma tabela, um offset de fluxo está localizado entre duas versões de tabela. A consulta de um fluxo retorna as alterações causadas por transações confirmadas após o offset e na hora atual ou antes dela.

O exemplo a seguir mostra uma tabela de origem com 10 versões confirmadas na linha do tempo. O offset do fluxo s1 está entre as versões de tabela v3 e v4. Quando o fluxo é consultado (ou consumido), os registros retornados incluem todas as transações entre a versão de tabela v4, a versão imediatamente após o offset do fluxo na linha de tempo da tabela, e v10, a versão de tabela mais recente confirmada na linha de tempo, inclusive.

Um fluxo fornece o conjunto mínimo de alterações desde o offset atual até a versão atual da tabela.

Múltiplas consultas podem consumir independentemente os mesmos dados de alteração de um fluxo sem alterar o offset. Um fluxo avança segundo o offset somente quando é usado em uma transação DML. Este comportamento se aplica tanto a transações explícitas como a transações de confirmação automática. (Por padrão, quando uma instrução DML é executada, uma transação de confirmação automática é implicitamente iniciada e a transação é confirmada na conclusão da instrução. Este comportamento é controlado com o parâmetro AUTOCOMMIT). Apenas consultar um fluxo não avança seu offset, mesmo dentro de uma transação explícita; o conteúdo do fluxo precisa ser consumido em uma instrução DML.

Quando uma instrução SQL consulta um fluxo dentro de uma transação explícita, o fluxo é consultado no ponto de avanço do fluxo (ou seja, o carimbo de data/hora) quando a transação começou, e não quando a instrução foi executada. Este comportamento diz respeito tanto a instruções DML como instruções CREATE TABLE … AS SELECT (CTAS) que preenchem uma nova tabela com linhas de um fluxo existente.

Uma instrução DML que seleciona de um fluxo consome todos os dados de alteração no fluxo, desde que a transação seja confirmada com sucesso. Para garantir que múltiplas instruções acessem os mesmos registros de alterações no fluxo, cerque-as com uma instrução de transação explícita (BEGIN … COMMIT). Isto bloqueia o fluxo. Atualizações DML do objeto de origem em transações paralelas são rastreadas pelo sistema de rastreamento de modificações, mas não atualizam o fluxo até que a instrução da transação explícita seja confirmada e os dados de alteração existentes sejam consumidos.

Isolamento de leitura repetível¶

Os fluxos suportam o isolamento de leitura repetível. No modo de leitura repetível, múltiplas instruções SQL dentro de uma transação veem o mesmo conjunto de registros em um fluxo. Isto difere do modo de leitura confirmada com suporte por tabelas, no qual as instruções veem quaisquer alterações feitas por instruções anteriores executadas dentro da mesma transação, mesmo que essas alterações ainda não estejam confirmadas.

Os registros delta retornados pelos fluxos em uma transação são o intervalo desde a posição atual do fluxo até a hora de início da transação. A posição do fluxo avança para a hora de início da transação se a transação for confirmada; caso contrário, ela permanece na mesma posição.

Considere o seguinte exemplo:

Dentro da Transação 1, todas as consultas do fluxo s1 veem o mesmo conjunto de registros. Alterações de DML na tabela t1 são registradas apenas no fluxo quando a transação é confirmada.

Na Transação 2, consultas ao fluxo veem as alterações registradas na tabela na Transação 1. Observe que se a Transação 2 tivesse começado antes da Transação 1 ter sido confirmada, consultas ao fluxo teriam retornado um instantâneo do fluxo desde a posição do fluxo até a hora de início da Transação 2 e não veriam nenhuma alteração confirmada pela Transação 1.

Colunas de fluxo¶

Um fluxo armazena um offset para o objeto de origem, e não quaisquer colunas ou dados reais da tabela. Quando consultado, um fluxo acessa e retorna os dados históricos na mesma forma que o objeto de origem (ou seja, os mesmos nomes de coluna e ordenação) com as seguintes colunas adicionais:

METADATA$ACTION:

Indica a operação DML (INSERT, DELETE) registrada.

METADATA$ISUPDATE:

Indica se a operação fazia parte de uma instrução UPDATE. As atualizações das linhas no objeto de origem são representadas como um par de registros DELETE e INSERT no fluxo com valores das coluna de metadados METADATA$ISUPDATE definidos como TRUE.

Observe que os fluxos registram as diferenças entre dois offsets. Se uma linha for adicionada e depois atualizada no offset atual, a alteração do delta é uma nova linha. A linha METADATA$ISUPDATE registra um valor FALSE.

METADATA$ROW_ID:

Especifica a ID única e imutável para a linha, que pode ser usada para rastrear alterações em linhas específicas ao longo do tempo.

Snowflake oferece as seguintes garantias em relação a METADATA$ROW_ID:

1. O METADATA$ROW_ID depende do objeto de origem do fluxo.

   Por exemplo, um fluxo stream1 na tabela table1 e um fluxo stream2 na tabela table1 produzem os mesmos METADATA$ROW_IDs para as mesmas linhas, mas um fluxo stream_view na exibição view1 não é garantido que produza os mesmos METADATA$ROW_IDs que stream1, mesmo que view seja definido usando a instrução CREATE VIEW view AS SELECT * FROM table1.

2. Um fluxo em um objeto de origem e um fluxo no clone do objeto de origem produzem os mesmos METADATA$ROW_IDs para as linhas que existem no momento da clonagem.

3. Um fluxo em um objeto de origem e um fluxo na réplica do objeto de origem produzem os mesmos METADATA$ROW_IDs para as linhas que foram replicadas.

Tipos de fluxos¶

Os seguintes tipos de fluxo estão disponíveis com base nos metadados registrados por cada um:

Padrão:

Com suporte para fluxos em tabelas, tabelas de diretório ou exibições. Um fluxo padrão (ou seja, delta) rastreia todas as alterações de DML no objeto de origem, incluindo inserções, atualizações e exclusões (incluindo truncamentos de tabela). Este tipo de fluxo realiza uma junção nas linhas inseridas e excluídas no conjunto de alterações para fornecer o delta de nível de linha. Como efeito líquido, por exemplo, uma linha que é inserida e depois excluída entre dois pontos de tempo transacionais em uma tabela é removida no delta (ou seja, não é retornada quando o fluxo é consultado).

Nota

Fluxos padrão não podem recuperar dados geoespaciais de alteração. Recomendamos a criação de fluxos apenas para anexação em objetos que contenham dados geoespaciais.

Apenas para anexação:

Com suporte para fluxos em tabelas, tabelas de diretório ou exibições padrão. Um fluxo apenas para anexação rastreia exclusivamente inserções de linha. As operações de atualização, exclusão e truncamento não são capturadas por fluxos apenas para anexação. Por exemplo, se 10 linhas forem inicialmente inseridas em uma tabela e, em seguida, 5 dessas linhas forem excluídas antes de avançar o deslocamento para um fluxo somente para anexação, o fluxo registrará apenas as 10 linhas inseridas.

Um fluxo somente para anexação retorna especificamente as linhas anexadas, tornando-o notavelmente mais eficiente do que um fluxo padrão para extração, carregamento e transformação (ELT) e cenários semelhantes que dependem exclusivamente de inserções de linha. Por exemplo, uma tabela de origem pode ser truncada imediatamente depois que as linhas de um fluxo apenas para anexação são consumidas, e as exclusões de registros não contribuirão para sobrecarga na próxima vez em que o fluxo for consultado ou consumido.

Não há suporte para a criação de fluxos somente para anexação em uma conta de destino usando um objeto secundário como origem.

Apenas para inserção:

Com suporte apenas para fluxos em tabelas externas. Um fluxo apenas para inserção rastreia apenas inserções de linhas; ele não registra operações de exclusão que removem linhas de um conjunto inserido (ou seja, sem operações). Por exemplo, entre quaisquer dois offsets, se Arquivo1 for removido do local de armazenamento em nuvem referenciado pela tabela externa, e Arquivo2 for adicionado, o fluxo só retorna registros para as linhas em Arquivo2. Diferente de quando se rastreia dados de CDC de tabelas padrão, Snowflake não pode acessar os registros históricos de arquivos em armazenamento em nuvem.

Os arquivos sobrescritos ou anexados são essencialmente tratados como novos arquivos: a versão antiga do arquivo é removida do armazenamento em nuvem, mas o fluxo apenas para inserção não registra a operação de exclusão. A nova versão do arquivo é adicionada ao armazenamento em nuvem, e o fluxo apenas para inserção registra as linhas como inserções. O fluxo não registra a diferença entre as versões antiga e nova do arquivo. Observe que os anexos podem não acionar uma atualização automática dos metadados de tabela externa, como quando se usa Azure AppendBlobs.

Fluxo de dados¶

O diagrama a seguir mostra como o conteúdo de um fluxo padrão muda conforme as linhas na tabela de origem são atualizadas. Sempre que uma instrução DML consome o conteúdo do fluxo, a posição do fluxo avança para acompanhar o próximo conjunto de alterações DML na tabela (ou seja, as alterações em uma versão da tabela):

Período de retenção de dados e desatualização¶

Um fluxo torna-se desatualizado quando seu offset está fora do período de retenção de dados para sua tabela de origem (ou para as tabelas subjacentes para uma exibição de origem). Quando um fluxo se torna desatualizado, os dados históricos para a tabela de origem não são mais acessíveis, incluindo quaisquer registros de alterações não consumidos. Para rastrear novos registros de alteração para a tabela, recrie o fluxo (usando CREATE STREAM). Para evitar que um fluxo se torne desatualizado, consuma os registros do fluxo dentro de uma transação durante o período de retenção para a tabela. Para obter mais informações sobre o período de retenção de dados, consulte Compreensão e uso do Time Travel.

Se o período de retenção de dados para uma tabela for inferior a 14 dias, e se um fluxo não tiver sido consumido, o Snowflake estende temporariamente este período para evitar que ele se torne desatualizado. O período é estendido para o offset do fluxo, até um máximo de 14 dias, por padrão, independentemente da edição do Snowflake para sua conta. O número máximo de dias para os quais o Snowflake pode estender o período de retenção de dados é determinado pelo valor do parâmetro MAX_DATA_EXTENSION_TIME_IN_DAYS. Quando o fluxo é consumido, o período de retenção de dados estendido é reduzido ao período padrão para a tabela.

A tabela seguinte mostra os valores de exemplo DATA_RETENTION_TIME_IN_DAYS e MAX_DATA_EXTENSION_TIME_IN_DAYS e indica com que frequência o conteúdo do fluxo deve ser consumido para evitar a desatualização:

Para ver o estado de desatualização atual de um fluxo, execute o comando DESCRIBE STREAM ou SHOW STREAMS. O carimbo de data/hora da coluna STALE_AFTER indica quando o fluxo está atualmente previsto para se tornar desatualizado (ou quando se tornou desatualizado, se o carimbo de data/hora estiver no passado). Este é o período de retenção de dados estendido para o objeto de origem. Este carimbo de data/hora é calculado adicionando o maior dos parâmetros DATA_RETENTION_TIME_IN_DAYS ou MAX_DATA_EXTENSION_TIME_IN_DAYS definidos para o objeto de origem do carimbo de data/hora atual.

Consumir os dados de alteração para um fluxo faz avançar o carimbo de data/hora STALE_AFTER. Note que a leitura do fluxo pode ser bem-sucedida por algum tempo após o carimbo de data/hora STALE_AFTER. Entretanto, o fluxo pode se tornar desatualizado a qualquer momento durante este período. A coluna STALE indica se é esperado que o fluxo esteja desatualizado, embora o fluxo possa ainda não estar.

Para evitar que um fluxo se torne desatualizado, recomendamos que você consuma regularmente os dados de alteração antes do carimbo de data/hora STALE_AFTER (ou seja, dentro do período de retenção de dados estendido para o objeto de origem).

Depois que o carimbo de data/hora STALE_AFTER tiver passado, o fluxo pode se tornar desatualizado a qualquer momento, mesmo que não haja registros não consumidos para o fluxo. Note que consultar um fluxo pode retornar 0 registros mesmo que haja dados de alteração para o objeto de origem. Por exemplo, um fluxo de dados apenas para anexação rastreia apenas inserções de linha, mas atualiza e exclui atividade e também grava registros de alteração em um objeto de origem. Um exemplo menos claro de gravações de tabela que não produzem dados de alteração é o reclustering.

O consumo dos dados de alteração para um fluxo avança seu offset para o presente, independentemente de as versões interpostas conterem ou não quaisquer dados de alteração.

Consumidores múltiplos de fluxos¶

Recomendamos que os usuários criem um fluxo separado para cada consumidor de registros de alterações para um objeto. “Consumidor” refere-se a uma tarefa, script ou outro mecanismo que consome os registros de alteração para um objeto usando uma transação DML. Como foi dito anteriormente neste tópico, um fluxo avança seu offset quando é utilizado em uma transação DML. Consumidores diferentes de dados de alteração em um único fluxo recuperam deltas diferentes, a menos que o Time Travel seja usado. Quando os dados de alteração capturados do último offset em um fluxo são consumidos usando uma transação DML, o fluxo avança segundo o offset. Os dados de alteração não estão mais disponíveis para o próximo consumidor. Para consumir os mesmos dados de alteração de um objeto, crie múltiplos fluxos para o objeto. Um fluxo armazena apenas um offset para o objeto de origem e não quaisquer dados reais da coluna da tabela; portanto, você pode criar qualquer número de fluxos para um objeto sem incorrer em custos significativos.

Streams on Views¶

Streams on Views oferecem suporte tanto a exibições locais quanto exibições compartilhadas usando o Snowflake Secure Data Sharing, incluindo exibições seguras. Atualmente, os fluxos não podem rastrear alterações em exibições materializadas.

Os fluxos são limitados a exibições que satisfazem os seguintes requisitos:

Tabelas subjacentes:

• Todas as tabelas subjacentes devem ser tabelas nativas.

• A exibição só pode aplicar as seguintes operações:

  • Projeções

  • Filtros

  • Junções internas ou cruzadas

  • UNION ALL

Exibições aninhadas e subconsultas na cláusula FROM têm suporte desde que a consulta totalmente expandida satisfaça os outros requisitos desta tabela de requisitos.

Consulta de exibição:

Requisitos gerais:

• A consulta pode selecionar qualquer número de colunas.

• A consulta pode conter qualquer número de predicados WHERE.

• Exibições com as seguintes operações ainda não têm suporte:

  • Cláusulas GROUP BY

  • Cláusulas QUALIFY

  • Subconsultas não incluídas na cláusula FROM.

  • Subconsultas correlacionadas

  • Cláusulas LIMIT

Funções:

• As funções na lista de seleção devem ser funções escalares definidas pelo sistema.

Rastreamento de alterações:

O rastreamento de alterações deve estar habilitado nas tabelas subjacentes.

Antes de criar um fluxo em uma exibição, você deve permitir o rastreamento de alterações nas tabelas subjacentes para a exibição. Para obter instruções, consulte Habilitação do rastreamento de alterações em exibições e tabelas subjacentes.

create or replace table orders (id int, order_name varchar);
create or replace table customers (id int, customer_name varchar);

create or replace view ordersByCustomer as select * from orders natural join customers;
insert into orders values (1, 'order1');
insert into customers values (1, 'customer1');

create or replace stream ordersByCustomerStream on view ordersBycustomer;

select * from ordersByCustomer;
+----+------------+---------------+
| ID | ORDER_NAME | CUSTOMER_NAME |
|----+------------+---------------|
|  1 | order1     | customer1     |
+----+------------+---------------+

select * exclude metadata$row_id from ordersByCustomerStream;
+----+------------+---------------+-----------------+-------------------+
| ID | ORDER_NAME | CUSTOMER_NAME | METADATA$ACTION | METADATA$ISUPDATE |
|----+------------+---------------+-----------------+-------------------|
+----+------------+---------------+-----------------+-------------------+

insert into orders values (1, 'order2');
select * from ordersByCustomer;
+----+------------+---------------+
| ID | ORDER_NAME | CUSTOMER_NAME |
|----+------------+---------------|
|  1 | order1     | customer1     |
|  1 | order2     | customer1     |
+----+------------+---------------+

select * exclude metadata$row_id from ordersByCustomerStream;
+----+------------+---------------+-----------------+-------------------+
| ID | ORDER_NAME | CUSTOMER_NAME | METADATA$ACTION | METADATA$ISUPDATE |
|----+------------+---------------+-----------------+-------------------|
|  1 | order2     | customer1     | INSERT          | False             |
+----+------------+---------------+-----------------+-------------------+

insert into customers values (1, 'customer2');
select * from ordersByCustomer;
+----+------------+---------------+
| ID | ORDER_NAME | CUSTOMER_NAME |
|----+------------+---------------|
|  1 | order1     | customer1     |
|  1 | order2     | customer1     |
|  1 | order1     | customer2     |
|  1 | order2     | customer2     |
+----+------------+---------------+

select * exclude metadata$row_id from ordersByCustomerStream;
+----+------------+---------------+-----------------+-------------------+
| ID | ORDER_NAME | CUSTOMER_NAME | METADATA$ACTION | METADATA$ISUPDATE |
|----+------------+---------------+-----------------+-------------------|
|  1 | order1     | customer2     | INSERT          | False             |
|  1 | order2     | customer1     | INSERT          | False             |
|  1 | order2     | customer2     | INSERT          | False             |
+----+------------+---------------+-----------------+-------------------+

Cláusula CHANGES: Alternativa apenas de leitura para fluxos¶

Como alternativa aos fluxos, o Snowflake suporta a consulta de metadados de rastreamento de alterações para tabelas ou exibições, usando a cláusula CHANGES para instruções SELECT. A cláusula CHANGES permite consultar os metadados de rastreamento de alterações entre dois pontos no tempo sem ter que criar um fluxo com um offset transacional explícito. Usar a cláusula CHANGES não avança o offset (ou seja, consome os registros). Consultas múltiplas podem recuperar os metadados de rastreamento de alterações entre diferentes pontos de extremidade iniciais e finais transacionais. Esta opção requer a especificação de um ponto inicial transacional para os metadados usando uma cláusula AT | BEFORE; o ponto final para o intervalo de rastreamento de alterações pode ser definido usando a cláusula opcional END.

Um fluxo armazena a versão de tabela transacional atual e é a origem apropriada de registros CDC na maioria dos cenários. Para cenários pouco frequentes que requerem o gerenciamento do offset por períodos de tempo arbitrários, a cláusula CHANGES está disponível para seu uso.

Atualmente, o seguinte deve ser verdadeiro para que os metadados de rastreamento de alterações sejam registrados:

Tabelas:

Habilite o rastreamento de alterações na tabela (usando ALTER TABLE … CHANGE_TRACKING = TRUE), ou crie um fluxo na tabela (usando CREATE STREAM).

Exibições:

Habilite o rastreamento de alterações na exibição e em suas tabelas subjacentes. Para obter instruções, consulte Habilitação do rastreamento de alterações em exibições e tabelas subjacentes.

O rastreamento de alterações adiciona várias colunas ocultas à tabela e começa a armazenar metadados de rastreamento de alterações. Os valores nestas colunas de dados CDC ocultos fornecem a entrada para as colunas de metadados do fluxo. As colunas consomem uma pequena quantidade de armazenamento.

Não há metadados de rastreamento de alterações para o objeto durante o período anterior ao cumprimento de uma dessas condições.

Privilégios de acesso requeridos¶

A consulta de um fluxo requer uma função com um mínimo das seguintes permissões de função:

Faturamento para fluxos¶

Como descrito em Período de retenção de dados e desatualização (neste tópico), quando um fluxo não é consumido regularmente, o Snowflake temporariamente estende o período de retenção de dados para a tabela de origem ou as tabelas subjacentes na exibição de origem. Se o período de retenção de dados para a tabela for inferior a 14 dias, então, nos bastidores, o período é estendido para o menor valor do offset do fluxo transacional offset ou 14 dias (se o período de retenção de dados para a tabela for inferior a 14 dias) independentemente da edição do Snowflake para sua conta.

A prorrogação do período de retenção de dados requer armazenamento adicional que será refletido em suas taxas mensais de armazenamento.

O principal custo associado a um fluxo é o tempo de processamento utilizado por um warehouse virtual para consultar o fluxo. Essas taxas aparecem em sua conta como créditos do Snowflake conhecidos.