Tabelas de streaming

Uma tabela de streaming é uma tabela Delta com suporte adicional para streaming ou processamento incremental de dados. Uma tabela de streaming pode ser alvo de um ou mais processos num pipeline de ETL.

As tabelas de streaming são uma boa opção para a ingestão de dados pelos seguintes motivos:

• Cada linha de entrada é processada apenas uma vez, o que representa a grande maioria das tarefas de ingestão (ou seja, acrescentando ou atualizando linhas numa tabela).
• Eles podem lidar com grandes volumes de dados em modo de acréscimo apenas.

As tabelas de streaming também são uma boa opção para transformações de streaming de baixa latência pelos seguintes motivos:

• Razão sobre linhas e janelas de tempo
• Lidar com grandes volumes de dados
• Baixa latência

O diagrama a seguir ilustra como as tabelas de streaming funcionam.

Em cada atualização, os fluxos associados a uma tabela de streaming leem as informações alteradas em uma fonte de streaming e acrescentam novas informações a essa tabela.

As tabelas de streaming são definidas e atualizadas por um único pipeline. Você define explicitamente tabelas de streaming no código-fonte do pipeline. As tabelas definidas por um pipeline não podem ser alteradas ou atualizadas por nenhum outro pipeline. Você pode definir vários fluxos para anexar a uma única tabela de streaming.

Quando você cria uma tabela de streaming fora de um pipeline no Databricks SQL, o Databricks cria um pipeline oculto que é usado para atualizar essa tabela.

Para mais informações sobre fluxos, consulte Carregamento e processamento de dados de forma incremental com fluxos de Pipelines Declarativos do Lakeflow.

Tabelas de fluxo contínuo para ingestão

As tabelas de streaming são projetadas para fontes de dados somente para adição e processam dados apenas uma vez.

O exemplo a seguir mostra como usar uma tabela de streaming para ingerir novos arquivos do armazenamento em nuvem.

Píton

import dlt

# create a streaming table
@dlt.table
def customers_bronze():
  return (
    spark.readStream.format("cloudFiles")
     .option("cloudFiles.format", "json")
     .option("cloudFiles.inferColumnTypes", "true")
     .load("/Volumes/path/to/files")
  )

Quando você usa a spark.readStream função em uma definição de conjunto de dados, ela faz com que Lakeflow Declarative Pipelines trate o conjunto de dados como um fluxo, e a tabela criada é uma tabela de streaming.

SQL

-- create a streaming table
CREATE OR REFRESH STREAMING TABLE customers_bronze
AS SELECT * FROM STREAM read_files(
  "/volumes/path/to/files",
  format => "json"
);

Para obter mais informações sobre como carregar dados numa tabela de streaming, consulte Carregar dados com Pipelines Declarativos do Lakeflow.

O diagrama a seguir ilustra como funcionam as tabelas de streaming com acréscimo único.

Uma linha que já tenha sido anexada a uma tabela de streaming não será consultada novamente com atualizações posteriores do pipeline. Se você modificar a consulta (por exemplo, de SELECT LOWER (name) para SELECT UPPER (name)), as linhas existentes não serão atualizadas para maiúsculas, mas as novas linhas serão maiúsculas. Você pode acionar uma atualização completa para requisitar novamente todos os dados anteriores da tabela de origem, atualizando todas as linhas na tabela de streaming.

Tabelas de streaming e streaming de baixa latência

As tabelas de streaming são projetadas para streaming de baixa latência em estado limitado. As tabelas de streaming usam o gerenciamento de pontos de verificação, o que as torna adequadas para streaming de baixa latência. No entanto, eles esperam fluxos que são naturalmente delimitados ou delimitados com uma marca d'água.

Um fluxo naturalmente limitado é produzido por uma fonte de dados de streaming que tem um início e um fim bem definidos. Um exemplo de um fluxo naturalmente limitado é a leitura de dados de um diretório de arquivos onde nenhum novo arquivo está sendo adicionado depois que um lote inicial de arquivos é colocado. O fluxo é considerado limitado porque o número de arquivos é finito e, em seguida, o fluxo termina depois que todos os arquivos foram processados.

Você também pode usar uma marca d'água para delimitar um fluxo. Uma marca d'água no Spark Structured Streaming é um mecanismo que ajuda a lidar com dados atrasados, especificando quanto tempo o sistema deve esperar por eventos atrasados antes de considerar a janela de tempo como concluída. Um fluxo não limitado sem marca d'água pode fazer com que um pipeline falhe devido à pressão da memória.

Para obter mais informações sobre o processamento de fluxo com estado, consulte Otimizar o processamento de fluxo com estado em Lakeflow Declarative Pipelines com marcas d'água.

Junção de stream-snapshot

As junções de instantâneo de fluxo são uniões entre um fluxo e uma dimensão que é capturada em instantâneo quando os fluxos são iniciados. Essas junções não serão recalculadas se a dimensão mudar após o início do fluxo, porque a tabela de dimensões é tratada como um instantâneo no tempo e as alterações na tabela de dimensões após o início do fluxo não são refletidas, a menos que você recarregue ou atualize a tabela de dimensões. Este é um comportamento razoável se puderes aceitar pequenas discrepâncias numa junção. Por exemplo, uma junção aproximada é aceitável quando o número de transações é muitas ordens de grandeza maior do que o número de clientes.

No exemplo de código a seguir, juntamos uma tabela de dimensão chamada "clientes", que possui duas linhas, com um conjunto de dados em constante crescimento, denominado "transações." Materializamos uma junção entre esses dois conjuntos de dados em uma tabela chamada sales_report. Observe que, se um processo externo atualizar a tabela de clientes adicionando uma nova linha (customer_id=3, name=Zoya), essa nova linha NÃO estará presente na junção porque a tabela de dimensão estática foi capturada em instantâneo quando os fluxos foram iniciados.

import dlt

@dlt.view
# assume this table contains an append-only stream of rows about transactions
# (customer_id=1, value=100)
# (customer_id=2, value=150)
# (customer_id=3, value=299)
# ... <and so on> ...
def v_transactions():
  return spark.readStream.table("transactions")

# assume this table contains only these two rows about customers
# (customer_id=1, name=Bilal)
# (customer_id=2, name=Olga)
@dlt.view
def v_customers():
  return spark.read.table("customers")

@dlt.table
def sales_report():
  facts = spark.readStream.table("v_transactions")
  dims = spark.read.table("v_customers")

  return (
    facts.join(dims, on="customer_id", how="inner"
  )

Limitações da tabela de streaming

As tabelas de streaming têm as seguintes limitações:

• Evolução limitada: você pode alterar a consulta sem recalcular todo o conjunto de dados. Como uma tabela de streaming só vê uma linha uma vez, você pode ter consultas diferentes operando em linhas diferentes. Isso significa que você deve estar ciente de todas as versões anteriores da consulta que estão sendo executadas em seu conjunto de dados. Uma atualização completa é necessária para fazer com que a tabela de streaming atualize os dados que já foram processados.
• gerenciamento de estado: As tabelas de streaming são de baixa latência, portanto, você precisa garantir que os fluxos sobre os quais operam sejam naturalmente limitados ou limitados com marca d'água. Para obter mais informações, consulte Otimizar o processamento com estado em Lakeflow Declarative Pipelines usando marcas d'água.
• As junções não são recalculadas: As junções em tabelas de streaming não são recalculadas quando as dimensões mudam. Esta característica pode ser boa para cenários "rápidos, mas errados". Se quiser que a sua vista esteja sempre correta, poderá querer utilizar uma vista materializada. As visualizações materializadas estão sempre corretas porque recalculam automaticamente as junções quando as dimensões mudam. Para obter mais informações, consulte Visões materializadas.