tabelas de transmissão

info

Visualização

Este recurso está em Visualização Pública.

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

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

Ao criar uma tabela de transmissão fora de um pipeline, usando Databricks SQL, Databricks cria um pipeline que é usado para atualizar a tabela. Você pode visualizar o pipeline selecionando "Tarefas e pipeline" na navegação à esquerda do seu workspace. Você pode adicionar a coluna do tipo de pipeline à sua view. As tabelas de transmissão definidas em um pipeline 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 Spark.

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")
  )

Para criar uma tabela de transmissão, a definição dataset deve ser do tipo transmissão. Quando você usa a função spark.readStream em uma definição dataset , ela retorna um dataset 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"
);

tabelas de transmissão requerem conjunto de dados de transmissão. A palavra-chave STREAM antes de read_files indica à consulta que o dataset deve ser tratado como uma transmissão.

Para obter mais detalhes sobre como carregar dados na tabela de transmissão, consulte Carregar dados no pipeline.

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ões com estado, consulte Otimizar o processamento com estado usando 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, portanto, é necessário garantir que as transmissões sobre as quais elas operam sejam naturalmente limitadas ou limitadas com marca d'água. Para obter mais informações, consulte Otimizar o processamento com estado usando 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.