tabelas de transmissão

Uma tabela de transmissão é uma tabela Delta com suporte adicional para transmissão ou processamento incremental de dados. Uma tabela de transmissão pode ser alvo de um ou mais fluxos em um pipeline ETL .

tabelas de transmissão são uma boa opção para aquisição de dados pelos seguintes motivos:

• Cada linha de entrada é manipulada apenas uma vez, o que modela a grande maioria das cargas de trabalho de ingestão (ou seja, anexando ou inserindo linhas em uma tabela).
• Eles podem lidar com grandes volumes de dados somente de acréscimo.

tabelas de transmissão também são uma boa escolha para transmissões 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 transmissão funcionam.

Em cada atualização, os fluxos associados a uma tabela de transmissão leem a informação alterada numa fonte de transmissão e acrescentam nova informação a essa tabela.

As tabelas de transmissão são definidas e atualizadas por um único pipeline. Você define explicitamente as tabelas de transmissão no código-fonte do pipeline. 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 transmissão.

nota

Quando você cria uma tabela de transmissão fora de um pipeline, usando Databricks SQL, Databricks cria um pipeline que é usado para atualizar a tabela. Você pode ver o pipeline selecionando Trabalhos e pipeline na navegação à esquerda no seu workspace. Você pode adicionar a coluna de tipo de pipeline à sua view. As tabelas de transmissão criadas no pipeline declarativo LakeFlow têm um tipo de ETL. As tabelas de transmissão criadas no Databricks SQL têm um tipo de MV/ST.

Para obter mais informações sobre fluxos, consulte Carregar e processar dados incrementalmente com fluxos de pipeline declarativos LakeFlow.

tabelas de transmissão para ingestão

As tabelas de transmissão são projetadas para somente anexar fonte de dados e processar entradas apenas uma vez.

O exemplo a seguir mostra como usar uma tabela de transmissão para ingerir novos arquivos do armazenamento cloud .

Python

Python

from pyspark import pipelines as dp

# create a streaming table
@dp.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 função spark.readStream em uma definição de dataset , isso faz com que o pipeline declarativo LakeFlow trate o dataset como uma transmissão, e a tabela criada é uma tabela de transmissão.

SQL

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 detalhes sobre o carregamento de dados na tabela de transmissão, consulte Carregar dados com o pipeline declarativo LakeFlow.

O diagrama a seguir ilustra como funcionam as tabelas de transmissão somente de acréscimo.

Uma linha que já foi anexada a uma tabela de transmissão não será consultada novamente com atualizações posteriores no 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 letras maiúsculas, mas as novas linhas serão maiúsculas. Você pode acionar uma refresh completa para consultar novamente todos os dados anteriores da tabela de origem para atualizar todas as linhas na tabela de transmissão.

tabelas de transmissão e transmissão de baixa latência

tabelas de transmissão são projetadas para transmissão de baixa latência em estado limitado. As tabelas de transmissão usam gerenciamento de ponto de verificação, o que as torna adequadas para transmissão de baixa latência. No entanto, eles esperam transmissões que sejam naturalmente delimitadas ou delimitadas com uma marca d'água.

Uma transmissão naturalmente delimitada é produzida por uma fonte de dados que tem início e fim bem definidos. Um exemplo de transmissão naturalmente limitada é a leitura de dados de um diretório de arquivos onde nenhum arquivo novo está sendo adicionado após um lote inicial de arquivos ser colocado. A transmissão é considerada limitada porque o número de arquivos é finito e, então, a transmissão termina depois que todos os arquivos foram processados.

Você também pode usar uma marca d'água para delimitar uma transmissão. Uma marca d'água no Spark transmissão estruturada é 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 completa. Uma transmissão ilimitada que não tem uma marca d'água pode causar falha em um pipeline devido à pressão de memória.

Para obter mais informações sobre o processamento de transmissão com estado, consulte Otimizar o processamento com estado no pipeline declarativo LakeFlow com marcas d'água.

transmissão-Junção instantânea

transmissão-Snapshot join é uma junção entre uma transmissão e uma dimensão que é capturada quando a transmissão começa. Essas junções não são recalculadas se a dimensão for alterada após o início da transmissão, porque a tabela de dimensões é tratada como um Snapshot no tempo, e as alterações na tabela de dimensões após o início da transmissão não são refletidas, a menos que você recarregue ou refresh a tabela de dimensões. Este é um comportamento razoável se você puder aceitar pequenas discrepâncias em uma join. Por exemplo, uma join aproximada é aceitável quando o número de transações é muitas ordens de magnitude maior que o número de clientes.

No exemplo de código a seguir, join uma tabela de dimensão, clientes, com duas linhas com um dataset cada vez maior, transações. Materializamos uma join 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 join porque a tabela de dimensão estática foi capturada quando as transmissões foram iniciadas.

Python

from pyspark import pipelines as dp

@dp.temporary_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)
@dp.temporary_view
def v_customers():
  return spark.read.table("customers")

@dp.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 transmissão

As tabelas de transmissão têm as seguintes limitações:

• Evolução limitada: você pode alterar a consulta sem recalcular todo o dataset. Sem uma refresh completa, uma tabela de transmissão só vê cada linha uma vez, então consultas diferentes terão processado linhas diferentes. Por exemplo, se você adicionar UPPER() a um campo na consulta, somente as linhas processadas após a alteração estarão em maiúsculas. Isso significa que você deve estar ciente de todas as versões anteriores da consulta que estão sendo executadas no seu dataset. Para reprocessar linhas existentes que foram processadas antes da alteração, é necessária uma refresh completa.
• Gerenciamento de estado: as tabelas de transmissão têm baixa latência, então você precisa garantir que a transmissão sobre a qual elas operam seja naturalmente limitada ou limitada com marca d'água. Para obter mais informações, consulte Otimizar o processamento com estado no pipeline declarativo LakeFlow com marcas d'água.
• join não recomputa: joins em tabelas de transmissão não recomputam quando as dimensões mudam. Essa característica pode ser boa para cenários “rápidos, mas errados”. Se você quiser que sua view esteja sempre correta, você pode usar uma view materializada. As visualizações materializadas estão sempre corretas porque elas recalculam automaticamente a junção quando as dimensões mudam. Para obter mais informações, consulte Visualização materializada.