Práticas recomendadas para pipelines do Lakeflow

Aplique esses padrões recomendados ao projetar, criar e operar pipelines, seja iniciando um novo pipeline ou melhorando um existente.

As páginas a seguir aprofundam decisões específicas de design:

Escolher o tipo de conjunto de dados correto

Os pipelines oferecem três tipos de conjunto de dados: tabelas de streaming, exibições materializadas e exibições temporárias. Escolher o tipo certo para cada camada do pipeline evita custos de computação desnecessários e mantém seu código fácil de raciocinar.

As tabelas de streaming são a opção certa para a ingestão de dados e transformações de streaming de baixa latência. Cada linha de entrada é lida e processada apenas uma vez, o que as torna ideais para workloads de acréscimo apenas, dados de alto volume e processamento acionado por eventos a partir de armazenamento em nuvem ou barramentos de mensagens.

Exibições materializadas são a escolha certa para transformações complexas e consultas analíticas. Seus resultados são previamente computados e mantidos atualizados usando a atualização incremental, portanto, as consultas em relação a eles são rápidas. Você não pode modificar diretamente os dados em uma exibição materializada; a definição de consulta controla a saída.

Exibições temporárias são exibições com escopo de pipeline que organizam sua lógica de transformação sem materializar dados no armazenamento. Use-as para etapas intermediárias que não precisam de sua própria tabela.

A tabela a seguir resume quando usar cada tipo:

Para obter mais informações, confira Tabelas de transmissão, Exibições materializadas e O que são os pipelines do Lakeflow?.

Utilize CDC declarativo em vez de MERGE imperativo

Implementar a CDC (captura de dados de alteração) com instruções SQL MERGE imperativas requer um código personalizado significativo para lidar corretamente com ordenação de eventos, eliminação de duplicação, atualizações parciais e evolução do esquema. Cada uma dessas preocupações deve ser resolvida de forma independente e o código resultante é difícil de manter e testar.

Pipelines fornecem a instrução AUTO CDC ... INTO (SQL) e a função create_auto_cdc_flow() (Python), que tratam de ordenação, eliminação de duplicatas, eventos fora de ordem e evolução do esquema de forma declarativa. Você descreve a estrutura do feed de alterações e da tabela de destino, e o pipeline cuida do resto. AUTO CDC dá suporte ao SCD Tipo 1 (substituição) e SCD Tipo 2 (preservação do histórico).

Para obter mais informações, consulte Alterar captura de dados e instantâneos e As APIS do AUTO CDC: Simplificar Alterar captura de dados com pipelines.

Reforçar a qualidade dos dados com as expectativas

As expectativas são expressões SQL verdadeiras/falsas aplicadas a cada linha que passa por um conjunto de dados. Quando uma linha falha na condição, o pipeline responde de acordo com a política de violação que você configurou. As expectativas emitem métricas para o log de eventos do pipeline, independentemente da política, para que você possa acompanhar as tendências de qualidade dos dados ao longo do tempo.

Selecionar uma política de infração

Três políticas de violação estão disponíveis. Escolha aquele que corresponda à tolerância a dados incorretos:

• aviso (padrão): os registros que não são válidos são gravados na tabela de destino e sinalizados em métricas. Use essa política quando precisar capturar todos os dados, mas quiser visibilidade dos problemas de qualidade.
• drop: Registros inválidos são descartados antes de serem gravados. Use isso quando linhas inválidas forem esperadas e não devem se propagar downstream.
• fail: a atualização do pipeline é interrompida no primeiro registro inválido. Use isso para dados críticos em que qualquer registro problemático indica um sério problema nas etapas anteriores.

Os exemplos a seguir mostram cada política aplicada a uma tabela de streaming:

SQL

-- Warn: write invalid records but track them in metrics
CREATE OR REFRESH STREAMING TABLE orders_raw (
  CONSTRAINT valid_order_id EXPECT (order_id IS NOT NULL)
) AS SELECT * FROM STREAM read_files("/volumes/raw/orders", format => "json");

-- Drop: discard invalid records before writing
CREATE OR REFRESH STREAMING TABLE orders_clean (
  CONSTRAINT non_negative_amount EXPECT (amount >= 0) ON VIOLATION DROP ROW
) AS SELECT * FROM STREAM(orders_raw);

-- Fail: stop the pipeline on any invalid record
CREATE OR REFRESH STREAMING TABLE orders_critical (
  CONSTRAINT required_customer_id EXPECT (customer_id IS NOT NULL) ON VIOLATION FAIL UPDATE
) AS SELECT * FROM STREAM(orders_clean);

Python

from pyspark import pipelines as dp

# Warn: write invalid records but track them in metrics
@dp.table
@dp.expect("valid_order_id", "order_id IS NOT NULL")
def orders_raw():
    return spark.readStream.format("cloudFiles") \
        .option("cloudFiles.format", "json") \
        .load("/volumes/raw/orders")

# Drop: discard invalid records before writing
@dp.table
@dp.expect_or_drop("non_negative_amount", "amount >= 0")
def orders_clean():
    return spark.readStream.table("orders_raw")

# Fail: stop the pipeline on any invalid record
@dp.table
@dp.expect_or_fail("required_customer_id", "customer_id IS NOT NULL")
def orders_critical():
    return spark.readStream.table("orders_clean")

Colocar registros inválidos em quarentena

Quando você quiser preservar os registros descartados para investigação, ao invés de descartá-los silenciosamente, use um padrão de quarentena. Encaminhe linhas que falham na validação para uma tabela de streaming separada usando dois fluxos: um que descarta linhas inválidas da tabela principal e um segundo que grava apenas as linhas inválidas em uma tabela de quarentena. Isso permite que você investigue, corrija e reprocesse dados incorretos sem contaminar seu conjunto de dados limpo.

Para obter um exemplo detalhado do padrão de quarentena, consulte recomendações de expectativa e padrões avançados.

Para obter mais informações sobre expectativas, consulte Gerenciar a qualidade dos dados com as expectativas do pipeline.

Parametrizar os pipelines

Os pipelines têm configurações padrão de catálogo e esquema, portanto, o código que lê e grava no mesmo catálogo e esquema funciona em ambientes sem parâmetros. No entanto, se o pipeline precisar fazer referência a um segundo catálogo ou esquema (por exemplo, a leitura de um catálogo de origem compartilhado que difere entre desenvolvimento e produção), evite codificar esses nomes diretamente em seu código-fonte. Em vez disso, defina-os como parâmetros de configuração de pipeline (pares chave-valor definidos nas configurações de pipeline) e referencie-os em seu código. Isso permite que uma única base de código seja executada corretamente entre ambientes trocando os valores de parâmetro.

SQL

CREATE OR REFRESH MATERIALIZED VIEW transaction_summary AS
SELECT account_id, COUNT(txn_id) AS txn_count, SUM(amount) AS total_amount
FROM ${source_catalog}.sales.transactions
GROUP BY account_id;

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import count, sum

@dp.materialized_view
def transaction_summary():
    source_catalog = spark.conf.get("source_catalog")
    return spark.read.table(f"{source_catalog}.sales.transactions") \
        .groupBy("account_id") \
        .agg(
            count("txn_id").alias("txn_count"),
            sum("amount").alias("total_amount")
        )

Para obter mais informações, consulte Como usar parâmetros com pipelines.

Escolha entre o modo de pipeline acionado e contínuo

O modo disparado processa todos os dados disponíveis e, em seguida, é interrompido. É a escolha certa para a grande maioria dos pipelines: aqueles que operam conforme um cronograma (por hora, diariamente ou sob demanda) e não exigem atualização de dados em menos de um minuto.

O modo contínuo mantém o cluster em execução e processa novos dados conforme ele chega. É apropriado somente quando seu caso de uso requer latência no intervalo de segundos a minutos. Como o modo contínuo requer um cluster sempre ativo, ele é significativamente mais caro do que o modo acionado.

O modo em tempo real se baseia no modo contínuo para alcançar latência inferior a um segundo, na faixa de milissegundos, para cargas de trabalho operacionais, como detecção de fraudes ou personalização em tempo real. Ele requer configuração adicional e planejamento de computação. Consulte Usar o modo em tempo real em pipelines do Lakeflow.

Para obter mais informações, consulte Modo de pipeline acionado vs. contínuo e Configurar pipelines.

Usar clustering do Liquid para layout de dados

O agrupamento dinâmico substitui o particionamento estático e o uso de ZORDER para otimizar o layout de dados em tabelas Delta. O particionamento estático exige que você selecione colunas de partição e reorganize os dados antecipadamente, o que pode causar distorção de dados para valores distribuídos de forma irregular. O clustering do Liquid é autoajustável, resistente a assimetrias e incremental, reescrevendo apenas os dados que precisam de reorganização em cada execução.

Altere as colunas de clustering a qualquer momento sem reescrever a tabela completa à medida que os padrões de consulta evoluem.

A Databricks recomenda o agrupamento líquido automático, que permite ao Databricks selecionar e manter as colunas de agrupamento ideais com base na carga de trabalho das consultas. Habilite-o com CLUSTER BY AUTO:

SQL

CREATE OR REFRESH STREAMING TABLE events
CLUSTER BY AUTO
AS SELECT * FROM STREAM read_files("/volumes/raw/events", format => "parquet");

Python

from pyspark import pipelines as dp

@dp.table(cluster_by_auto=True)
def events():
    return spark.readStream.format("cloudFiles") \
        .option("cloudFiles.format", "parquet") \
        .load("/volumes/raw/events")

Para escolher as colunas de clustering por conta própria, especifique-as explicitamente:

SQL

CREATE OR REFRESH STREAMING TABLE events
CLUSTER BY (event_date, region)
AS SELECT * FROM STREAM read_files("/volumes/raw/events", format => "parquet");

Python

from pyspark import pipelines as dp

@dp.table(cluster_by=["event_date", "region"])
def events():
    return spark.readStream.format("cloudFiles") \
        .option("cloudFiles.format", "parquet") \
        .load("/volumes/raw/events")

Para obter mais informações, consulte as Tabelas de streaming e Usar o clustering do Liquid para tabelas.

Gerenciar pipelines com CI/CD e Pacotes de Automação Declarativa

Gerencie o controle de versão do código-fonte do pipeline e utilize Pacotes Declarativos de Automação para administrar implantações em diferentes ambientes.

Para obter mais informações, consulte Criar um pipeline controlado pela origem, Converter um pipeline em um projeto de pacote e Usar parâmetros com pipelines.

Armazenar código de fluxo de trabalho no sistema de controle de versão

Armazene todos os arquivos de origem do pipeline (Python e SQL) junto com a configuração do pacote em um repositório Git. O controle de versão do projeto completo oferece um histórico completo de alterações, facilita a colaboração e permite que você valide as alterações em um ambiente de desenvolvimento antes de promovê-las à produção.

O Databricks recomenda Pacotes de Automação Declarativa para gerenciar esse fluxo de trabalho. Um pacote define sua configuração de pipeline no YAML junto com o código-fonte e a databricks bundle CLI permite validar, implantar e executar pipelines do terminal ou de um sistema de CI/CD.

Usar destinos de pacote para isolamento de ambiente

Os pacotes permitem vários alvos (por exemplo, dev, staging, prod), cada um com seu próprio conjunto de sobreposições para nomes de catálogo, políticas de cluster, endereços de notificação e outras configurações. Combine os alvos de pacote com parâmetros de pipeline para injetar os valores corretos específicos do ambiente no momento da implantação, mantendo seu código-fonte livre de constantes de configuração de ambiente.

Um fluxo de trabalho típico tem esta aparência:

1. Um desenvolvedor trabalha em uma ramificação de recursos, implantando em um pipeline de desenvolvimento individual em um catálogo de desenvolvimento do desenvolvedor.
2. Ao mesclar a ramificação principal, um sistema de CI executa databricks bundle validate e databricks bundle deploy --target staging para validar e implantar o pipeline em um ambiente de preparo.
3. Após a conclusão dos testes, o sistema de Integração Contínua (CI) é implantado em produção com databricks bundle deploy --target prod.

Práticas recomendadas de streaming

Use esses padrões para gerenciar o estado, controlar dados atrasados e manter os pipelines de streaming confiáveis.

Para obter mais informações, consulte Otimizar o processamento com estado mantido utilizando marcas d'água, Recuperar um pipeline após falha de ponto de verificação em streaming e Provisionamento de dados históricos com pipelines.

Usar marcas d'água para operações com estado

As marcas d'água delimitam o estado que o pipeline mantém na memória durante operações de streaming com estado, como agregações em janelas e desduplicação. Sem uma marca d'água, o estado cresce sem limites à medida que o pipeline acumula dados para cada chave possível, eventualmente causando erros de falta de memória em pipelines de longa execução.

Uma marca d'água especifica uma coluna de carimbo de data/hora e um limite de tolerância para dados recebidos com atraso. Os registros que chegam depois que o limite é passado são descartados. Escolha um limite que balancee sua tolerância a dados atrasados em relação ao custo de memória de manter esse estado aberto.

O exemplo a seguir calcula uma agregação de janela em cascata de um minuto com uma marca d'água de três minutos.

SQL

CREATE OR REFRESH STREAMING TABLE event_counts AS
SELECT window(event_time, '1 minute') AS time_window, region, COUNT(*) AS cnt
FROM STREAM(events_raw)
  WATERMARK event_time DELAY OF INTERVAL 3 MINUTES
GROUP BY time_window, region;

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import window

@dp.table
def event_counts():
    return (
        spark.readStream.table("events_raw")
            .withWatermark("event_time", "3 minutes")
            .groupBy(window("event_time", "1 minute"), "region")
            .count()
    )

Entender o estado de streaming e a atualização completa

O estado de streaming é incremental: o pipeline constrói e mantém o estado através das atualizações, em vez de recalcular do zero a cada vez. Isso é o que torna o streaming com estado eficiente, mas também significa que se você alterar a lógica de uma consulta com estado (por exemplo, modificar um limite de marca d'água ou alterar colunas de agregação), o estado existente não será mais compatível com a nova lógica. Nesse caso, você deve executar uma atualização completa para reprocessar todos os dados históricos com a nova lógica e recompilar o estado do zero.

Uma atualização completa também poderá levar à perda de dados se a origem não reter dados históricos. Por exemplo, uma fonte kafka com um curto período de retenção pode ter apenas os últimos minutos de dados disponíveis no momento da atualização, resultando em uma tabela que contém muito menos dados do que antes. Planeje alterações de lógica de consulta com estado cuidadosamente, especialmente para fluxos de alto volume em que uma atualização completa é cara ou em que a fonte tem retenção de dados limitada. O uso da arquitetura de medalhão ajuda ao criar tabelas bronze com transformação mínima e permite que as tabelas prata ou ouro sejam recalculadas a partir das tabelas bronze mantendo todo o histórico de dados.

Junções de fluxo a fluxo

As junções de fluxo a fluxo exigem uma marca d'água em ambos os lados da junção e uma condição de junção com limitação do intervalo. O intervalo de tempo na condição de junção informa ao mecanismo de streaming quando não há mais correspondências possíveis, permitindo que ele remova o estado que não pode mais ser correspondido. Se você omitir as marcas-d’água ou a condição de limite de tempo, o estado crescerá sem limites.

O exemplo a seguir une eventos de impressão de anúncios com eventos de clique, exigindo que o clique ocorra dentro de três minutos após a impressão:

SQL

CREATE OR REFRESH STREAMING TABLE impression_clicks AS
SELECT imp.ad_id, imp.impression_time, clk.click_time
FROM STREAM(ad_impressions)
    WATERMARK impression_time DELAY OF INTERVAL 3 MINUTES AS imp
JOIN STREAM(user_clicks)
    WATERMARK click_time DELAY OF INTERVAL 3 MINUTES AS clk
ON imp.ad_id = clk.ad_id
  AND clk.click_time BETWEEN imp.impression_time
    AND imp.impression_time + INTERVAL 3 MINUTES;

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import expr

dp.create_streaming_table("impression_clicks")

@dp.append_flow(target="impression_clicks")
def join_impressions_and_clicks():
    impressions = spark.readStream.table("ad_impressions") \
        .withWatermark("impression_time", "3 minutes")
    clicks = spark.readStream.table("user_clicks") \
        .withWatermark("click_time", "3 minutes")
    return impressions.alias("imp").join(
        clicks.alias("clk"),
        expr("""
            imp.ad_id = clk.ad_id AND
            clk.click_time BETWEEN imp.impression_time AND imp.impression_time + INTERVAL 3 MINUTES
        """),
        "leftOuter"
    )

Quando você combina um fluxo com uma tabela estática (uma junção do tipo instantâneo), o instantâneo da tabela estática é atualizado novamente no início de cada microlote. Isso significa que os registros de dimensões que chegam tardiamente não são aplicados retroativamente a fatos que já foram processados. Se for necessária a aplicação retroativa, use uma visão materializada ou reestruture o pipeline.

Otimizar o desempenho do pipeline

Aplique essas técnicas para reduzir os custos de computação e acelerar as atualizações de pipeline.

Para obter mais informações, consulte Exibições materializadas e Otimizar o processamento com estado utilizando marcas d'água.

Evitar arquivos pequenos

Acionar um pipeline frequentemente em uma fonte de baixa taxa de dados escreve um grande número de arquivos pequenos no armazenamento em nuvem. Arquivos pequenos degradam o desempenho de leitura porque cada arquivo requer uma pesquisa de metadados separada e uma viagem de ida e volta de E/S, e as APIs de armazenamento em nuvem limitam as operações de listagem em escala. Para evitar isso, escolha um intervalo de gatilho que corresponda ao volume de dados: execute pipelines disparados em um agendamento que permita que uma quantidade significativa de dados se acumule entre atualizações, em vez de continuamente.

Manipular distorção de dados

A distorção de dados ocorre quando os valores em uma chave join ou groupBy são distribuídos de forma desigual entre partições, fazendo com que um pequeno número de tarefas processe a maioria dos dados. Isso cria hotspots que aumentam o tempo de atualização de ponta a ponta. Use o clustering do Liquid para solucionar distorção em tabelas armazenadas. Para lidar com a distribuição desigual de dados que ocorre durante a computação em andamento, distribua as chaves com alta distribuição desigual, acrescentando um sufixo aleatório de bucket antes de agrupar e realizar a agregação em duas etapas.

Para obter mais informações, consulte Utilizar agrupamento líquido para a disposição dos dados.

Usar atualização incremental para exibições materializadas

Quando você usa uma exibição materializada para uma agregação de grande porte, o pipeline tenta atualizá-la incrementalmente, processando apenas as alterações nos dados de origem desde a última atualização, em vez de recomputar o conjunto completo de resultados. A atualização incremental é significativamente mais barata do que reexecutar a consulta desde o início em cada gatilho de pipeline. Para maximizar a chance de que uma exibição materializada possa ser atualizada incrementalmente, escreva consultas de agregação simples e determinísticas e evite construções que impeçam o processamento incremental, como funções não determinísticas.

Confira Atualização incremental para exibições materializadas.

Otimizar junções

Para junções em que um lado é uma pequena tabela de dimensões, adicione uma dica de difusão para instruir o Spark a transmitir a tabela menor para todos os executores em vez de executar uma junção aleatória:

SQL

CREATE OR REFRESH MATERIALIZED VIEW enriched_orders AS
SELECT o.*, /*+ BROADCAST(p) */ p.product_name, p.category
FROM orders o
JOIN products p ON o.product_id = p.product_id;

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import broadcast

@dp.materialized_view
def enriched_orders():
    orders = spark.read.table("orders")
    products = spark.read.table("products")
    return orders.join(broadcast(products), "product_id")

Para junções por proximidade em séries temporais (por exemplo, encontrar o evento mais próximo dentro de um intervalo de tempo), use uma condição de junção de intervalo e verifique se ambos os lados têm uma marca d'água se ingressarem em fluxos, ou considere pré-compartimentalização de eventos em intervalos de tempo antes de ingressar.

Monitorar os pipelines

O log de eventos do pipeline é o principal recurso de observabilidade nos pipelines. Cada execução de um pipeline grava registros estruturados no log de eventos, abrangendo o progresso da execução, os resultados das verificações de qualidade dos dados, a proveniência dos dados e os detalhes do erro. O log de eventos é uma tabela Delta que você pode consultar diretamente.

Para consultar o log de eventos sem conhecer o caminho de armazenamento subjacente, use a event_log() função com valor de tabela em um cluster compartilhado ou SQL warehouse:

SELECT * FROM event_log('<pipeline-id>')
WHERE event_type = 'flow_progress'
ORDER BY timestamp DESC
LIMIT 100;

Crie painéis de qualidade de dados consultando o log de eventos para as métricas de expectativa. A details coluna contém uma estrutura JSON aninhada com contagens de aprovação/falha para cada restrição, que você pode usar para acompanhar tendências de qualidade ao longo do tempo e alertar sobre regressões.

Para alertas controlados por eventos, use ganchos de evento para disparar webhooks personalizados ou serviços de notificação (como Slack ou PagerDuty) quando um pipeline falha ou quando um limite de qualidade de dados é violado. Ganchos de evento são funções do Python que são executadas em resposta a eventos de pipeline.

Para obter mais informações, consulte Monitoramento de pipelines, Log de eventos de pipelines e Definir monitoramento personalizado para pipelines com ganchos de eventos.

Usar computação sem servidor

O Databricks recomenda a computação sem servidor para novos pipelines. Com o serverless, não há nenhuma configuração manual de cluster; O Databricks gerencia a infraestrutura automaticamente. Os pipelines sem servidor usam dimensionamento automático aprimorado que pode dimensionar horizontalmente (mais executores) e verticalmente (tamanho maior do executor) em resposta às demandas de carga de trabalho. Os pipelines sem servidor sempre usam o Catálogo do Unity, portanto, a governança e o acompanhamento de linhagem são integrados por padrão.

Serverless também é necessário para uma atualização incremental das visualizações materializadas. Na computação clássica, as visualizações materializadas são sempre totalmente recalculadas, o que aumenta os custos de atualização. Se a atualização incremental for importante para sua carga de trabalho, use computação serverless.

Para uma comparação entre computação serverless e computação clássica, consulte Computação serverless vs. computação clássica para pipelines. Para mais informações sobre serverless, consulte Configurar um pipeline sem servidor.

Organizar pipelines com a arquitetura de medalhão

A arquitetura do medalhão organiza dados em três camadas lógicas (bronze, prata e ouro), cada uma com uma finalidade distinta. Mapear os tipos de conjuntos de dados do pipeline à camada correta mantém claras as responsabilidades de cada camada e torna os pipelines mais fáceis de manter.

• Bronze: Use tabelas de streaming em tempo real para ingerir dados brutos do armazenamento em nuvem, barramentos de mensagens ou fontes de CDC. As tabelas bronze preservam os dados brutos de origem com transformação mínima, possibilitando que camadas prata ou ouro sejam reprocessadas da origem na camada bronze se os requisitos forem alterados.
• Prata: Use tabelas de streaming para transformações incrementais no nível de linha (filtragem, limpeza e análise sintática). Use exibições materializadas quando a lógica da camada prata envolve junções de enriquecimento em tabelas de dimensão ou agregações complexas que se beneficiam da atualização incremental.
• Ouro: use exibições materializadas para calcular previamente agregações, métricas e resumos servidos a painéis, ferramentas de relatórios e consumidores downstream.

Separe a ingestão (bronze) e a transformação (prata e ouro) em pipelines distintos sempre que possível. Desacoplar as camadas permite agendar, monitorar e diagnosticar cada camada independentemente, e uma falha em um pipeline de transformação não impede que novos dados cheguem ao nível bronze.

Para obter mais informações, consulte Tabelas de streaming e Exibições materializadas.