Provisionamento de dados históricos com pipelines

Na engenharia de dados, provisionamento refere-se ao processo de processamento retroativo de dados históricos por meio de um pipeline de dados que foi projetado para processar dados em tempo real ou de streaming.

Normalmente, esse é um fluxo separado que envia dados para suas tabelas existentes. A ilustração a seguir mostra um fluxo de provisionamento retroativo enviando dados históricos para as tabelas bronze no seu pipeline.

Alguns cenários que podem exigir uma reposição de dados:

• Processe dados históricos de um sistema herdado para treinar um modelo de ML (machine learning) ou criar um painel de análise de tendências históricas.
• Reprocesse um subconjunto de dados devido a um problema de qualidade de dados com fontes de dados upstream.
• Seus requisitos de negócios foram alterados e você precisa fazer o backup de dados para um período de tempo diferente que não foi coberto pelo pipeline inicial.
• Sua lógica de negócios mudou e você precisa reprocessar dados históricos e atuais.

O fluxo de preenchimento que você usa depende da tabela alvo e dos dados fonte. Para um AUTO CDC alvo Tipo 1 de dimensão que muda lentamente (SCD) com um snapshot autoritativo, use um fluxo único AUTO CDC FROM SNAPSHOT . Para uma migração SCD que reproduz mudanças históricas, use um fluxo único AUTO CDC .

Para backfills somente de acréscimo: Use um fluxo de acréscimo especializado com a opção ONCE para fazer o backfill de uma tabela de streaming somente de acréscimo. Consulte append_flow ou CREATE FLOW (pipelines) para obter mais informações sobre a opção ONCE .

Considerações ao fazer backup de dados históricos em uma tabela de streaming

• Normalmente, acrescente os dados à tabela de streaming bronze. As camadas de prata e ouro a jusante captam os novos dados da camada de bronze.
• Verifique se o pipeline pode lidar com dados duplicados normalmente caso os mesmos dados sejam acrescentados várias vezes.
• Verifique se o esquema de dados históricos é compatível com o esquema de dados atual.
• Considere o tamanho do volume de dados e o acordo de nível de serviço (SLA) de processamento necessário, e configure os tamanhos do cluster e do lote.

Exemplo: Adicionar um backfill a um pipeline existente

Neste exemplo, digamos que você tenha um pipeline que ingere dados brutos de registro de eventos de uma fonte de armazenamento em nuvem, a partir de 01 de janeiro de 2025. Depois você percebe que quer preencher retroativamente os três anos anteriores de dados históricos para casos de uso de relatórios e análises downstream. Todos os dados estão em um local, particionados por ano, mês e dia, no formato JSON.

Pipeline inicial

Aqui está o código de pipeline inicial que ingere incrementalmente os dados brutos de registro de eventos do armazenamento em nuvem.

Python

from pyspark import pipelines as dp

source_root_path = spark.conf.get("registration_events_source_root_path")
begin_year = spark.conf.get("begin_year")
incremental_load_path = f"{source_root_path}/*/*/*"

# create a streaming table and the default flow to ingest streaming events
@dp.table(name="registration_events_raw", comment="Raw registration events")
def ingest():
    return (
        spark
        .readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.inferColumnTypes", "true")
        .option("cloudFiles.maxFilesPerTrigger", 100)
        .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
        .option("modifiedAfter", "2025-01-01T00:00:00.000+00:00")
        .load(incremental_load_path)
        .where(f"year(timestamp) >= {begin_year}") # safeguard to not process data before begin_year
    )

SQL

-- create a streaming table and the default flow to ingest streaming events
CREATE OR REFRESH STREAMING LIVE TABLE registration_events_raw AS
SELECT * FROM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/*/*/*",
  format => "json",
  inferColumnTypes => true,
  maxFilesPerTrigger => 100,
  schemaEvolutionMode => "addNewColumns",
  modifiedAfter => "2024-12-31T23:59:59.999+00:00"
)
WHERE year(timestamp) >= '2025'; -- safeguard to not process data before begin_year

Aqui, usamos a opção modifiedAfter Carregador Automático para garantir que não estamos processando todos os dados do caminho de armazenamento em nuvem. O processamento incremental é cortado nesse limite.

Dados de preenchimento retrospectivo dos três anos anteriores

Agora você deseja adicionar um ou mais fluxos para fazer backup de dados anteriores. Neste exemplo, execute as seguintes etapas:

• Utilize o append once fluxo. Isso realiza o provisionamento de uma única vez e não continua sendo executado após esse primeiro provisionamento. O código permanece em seu pipeline e, se ele for totalmente atualizado, o provisionamento será executado novamente.
• Crie três fluxos de provisionamento, um para cada ano (nesse caso, os dados são divididos por ano no caminho). Para Python, parametrizamos a criação dos fluxos, mas no SQL repetimos o código três vezes, uma vez para cada fluxo.

Se você estiver trabalhando em seu próprio projeto e não estiver usando computação sem servidor, talvez queira atualizar o número máximo de trabalhadores para o pipeline. Aumentar o número máximo de trabalhadores garante que você tenha os recursos para processar os dados históricos enquanto continua a processar os dados de streaming atuais dentro do SLA previsto.

Python

from pyspark import pipelines as dp

source_root_path = spark.conf.get("registration_events_source_root_path")
begin_year = spark.conf.get("begin_year")
backfill_years = spark.conf.get("backfill_years") # e.g. "2024,2023,2022"
incremental_load_path = f"{source_root_path}/*/*/*"

# meta programming to create append once flow for a given year (called later)
def setup_backfill_flow(year):
    backfill_path = f"{source_root_path}/year={year}/*/*"
    @dp.append_flow(
        target="registration_events_raw",
        once=True,
        name=f"flow_registration_events_raw_backfill_{year}",
        comment=f"Backfill {year} Raw registration events")
    def backfill():
        return (
            spark
            .read
            .format("json")
            .option("inferSchema", "true")
            .load(backfill_path)
        )

# create the streaming table
dp.create_streaming_table(name="registration_events_raw", comment="Raw registration events")

# append the original incremental, streaming flow
@dp.append_flow(
        target="registration_events_raw",
        name="flow_registration_events_raw_incremental",
        comment="Raw registration events")
def ingest():
    return (
        spark
        .readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.inferColumnTypes", "true")
        .option("cloudFiles.maxFilesPerTrigger", 100)
        .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
        .option("modifiedAfter", "2024-12-31T23:59:59.999+00:00")
        .load(incremental_load_path)
        .where(f"year(timestamp) >= {begin_year}")
    )

# parallelize one time multi years backfill for faster processing
# split backfill_years into array
for year in backfill_years.split(","):
    setup_backfill_flow(year) # call the previously defined append_flow for each year

SQL

-- create the streaming table
CREATE OR REFRESH STREAMING TABLE registration_events_raw;

-- append the original incremental, streaming flow
CREATE FLOW
  registration_events_raw_incremental
AS INSERT INTO
  registration_events_raw BY NAME
SELECT * FROM STREAM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/*/*/*",
  format => "json",
  inferColumnTypes => true,
  maxFilesPerTrigger => 100,
  schemaEvolutionMode => "addNewColumns",
  modifiedAfter => "2024-12-31T23:59:59.999+00:00"
)
WHERE year(timestamp) >= '2025';


-- one time backfill 2024
CREATE FLOW
  registration_events_raw_backfill_2024
AS INSERT INTO ONCE
  registration_events_raw BY NAME
SELECT * FROM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/year=2024/*/*",
  format => "json",
  inferColumnTypes => true
);

-- one time backfill 2023
CREATE FLOW
  registration_events_raw_backfill_2023
AS INSERT INTO ONCE
  registration_events_raw BY NAME
SELECT * FROM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/year=2023/*/*",
  format => "json",
  inferColumnTypes => true
);

-- one time backfill 2022
CREATE FLOW
  registration_events_raw_backfill_2022
AS INSERT INTO ONCE
  registration_events_raw BY NAME
SELECT * FROM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/year=2022/*/*",
  format => "json",
  inferColumnTypes => true
);

Essa implementação realça vários padrões importantes.

Separação de interesses

• O processamento incremental é independente das operações de backfill.
• Cada fluxo tem suas próprias configurações e configurações de otimização.
• Há uma clara distinção entre operações incrementais e de backfill.

Execução controlada

• Usar a opção ONCE garante que cada backfill seja executado exatamente uma vez.
• O fluxo de provisionamento permanece no grafo de pipeline, mas fica ocioso após sua conclusão. Ele está pronto para uso na atualização completa, automaticamente.
• Há uma trilha de auditoria clara das operações de provisionamento retroativo na definição do pipeline.

Otimização de processamento

• Você pode dividir o provisionamento grande em vários provisionamentos menores para processamento mais rápido ou para controle sobre o processamento.
• O dimensionamento automático aprimorado ajusta dinamicamente o tamanho do cluster com base na carga atual do cluster.

Evolução do esquema

• Usar schemaEvolutionMode="addNewColumns" lida com as alterações de esquema de forma eficiente.
• Você tem inferência de esquema consistente entre dados históricos e atuais.
• Há uma manipulação segura de novas colunas em dados mais recentes.

Adicionar um backfill a uma tabela AUTO CDC SCD Tipo 1

Use um fluxo único AUTO CDC FROM SNAPSHOT para adicionar um snapshot autoritativo a um alvo SCD Tipo 1 que também receba um feed contínuo de captura de dados de mudança (CDC). A versão snapshot e a coluna de sequenciamento CDC formam um domínio de ordenação. Um evento CDC mais recente tem precedência sobre um snapshot mais antigo, enquanto um snapshot mais recente tem precedência sobre um evento CDC mais antigo.

Requisitos

Antes de adicionar o preenchimento, certifique-se de que os fluxos atendam aos seguintes requisitos:

• O destino utiliza SCD Tipo 1.
• O alvo possui exatamente um AUTO CDC FROM SNAPSHOT fluxo e um ou mais fluxos AUTO CDC com nomes exclusivos.
• Todos os fluxos usam o mesmo número de chaves na mesma ordem. Nomes de chaves snapshot-flow são comparados sem diferenciar maiúsculas de minúsculas com AUTO CDC nomes de chaves. Múltiplos AUTO CDC fluxos devem usar nomes-chave idênticos e o mesmo uso de maiúsculas e minúsculas.
• A versão snapshot e cada coluna de sequenciamento CDC têm exatamente o mesmo tipo de dado.
• O AUTO CDC FROM SNAPSHOT fluxo não define expectativas.
• Os AUTO CDC fluxos não usam IGNORE NULL UPDATES. Em Python, não defina ignore_null_updates, ignore_null_updates_column_list, ou ignore_null_updates_except_column_list.
• O pipeline usa modo acionado. Esse padrão não suporta pipelines contínuos.

Ambos os tipos de fluxo podem usar a interface de pipeline SQL ou Python. Você pode misturar fluxos SQL e Python no mesmo destino.

O snapshot deve representar o estado completo da origem em sua versão. Se uma chave de destino estiver ausente do snapshot, AUTO CDC FROM SNAPSHOT trata a ausência como uma exclusão na versão do snapshot. Um evento do CDC com uma versão mais recente preserva ou restaura a chave.

Adicione o backfill

Para adicionar um backfill de snapshot único e continuar processando eventos CDC, utilize os seguintes passos:

1. Mantenha a tabela de destino existente e seus fluxos em andamento AUTO CDC na definição do pipeline.
2. Defina o snapshot autoritativo e sua versão. Para um callback em Python, a primeira invocação deve retornar um snapshot e uma versão. Devolva None somente após pelo menos um snapshot ter sido processado.
3. Adicione um AUTO CDC FROM SNAPSHOT fluxo com once=True em Python ou ONCE em SQL. Para um backfill SQL em um destino existente, inclua uma WITH VERSION consulta. Um fluxo snapshot SQL sem WITH VERSION suporta apenas uma carga inicial em um destino vazio.
4. Execute uma atualização acionada do pipeline para processar o backfill e os eventos contínuos de CDC.

O exemplo a seguir começa com um pipeline Python existente que processa incrementalmente as alterações em customers_cdc. Suponha que você já tenha rodado esse pipeline e preenchido o customers target:

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

@dp.view
def customers_cdc():
    return (
        spark.readStream.table("main.bronze.customers_cdc")
        .withColumn("change_timestamp", col("change_timestamp").cast("timestamp"))
    )


dp.create_streaming_table("customers")

dp.create_auto_cdc_flow(
    name="customers_incremental_cdc",
    target="customers",
    source="customers_cdc",
    keys=["customer_id"],
    sequence_by=col("change_timestamp"),
    apply_as_deletes=expr("operation = 'DELETE'"),
    except_column_list=["operation", "change_timestamp"],
    stored_as_scd_type=1,
)

Para preencher esse destino existente com o estado de customers_snapshot em 1º de janeiro de 2025, adicione o código a seguir à mesma definição de pipeline. Mantenha a tabela de destino existente e o fluxo AUTO CDC:

from datetime import datetime, timezone
from typing import Optional, Tuple

from pyspark.sql import DataFrame

backfill_version = datetime(2025, 1, 1, tzinfo=timezone.utc)


def backfill_snapshot_and_version(
    latest_snapshot_version: Optional[datetime],
) -> Optional[Tuple[DataFrame, datetime]]:
    if latest_snapshot_version is None:
        return (spark.read.table("main.legacy.customers_snapshot"), backfill_version)
    return None


dp.create_auto_cdc_from_snapshot_flow(
    target="customers",
    source=backfill_snapshot_and_version,
    keys=["customer_id"],
    stored_as_scd_type=1,
    once=True,
)

O callback deve retornar um snapshot e uma versão em sua primeira invocação. Se retornar None antes de qualquer snapshot ser processado, a atualização do pipeline falha. Após o processamento do snapshot, retornar None sinaliza que não há snapshots adicionais disponíveis.

A versão snapshot é um objeto Python datetime, que corresponde ao tipo Spark SQL TIMESTAMP. O fluxo existente AUTO CDC converte change_timestamp em TIMESTAMP para que os dois tipos de sequenciamento correspondam exatamente. O exemplo usa Python para ambos os fluxos, mas você pode definir qualquer um dos fluxos em SQL e misturar fluxos SQL e Python no mesmo destino. Para a sintaxe SQL, incluindo a consulta obrigatória WITH VERSION para um destino não vazio, veja CREATE FLOW (pipelines).

Após o fluxo de snapshot ser confirmado com sucesso, atualizações incrementais subsequentes o pulam enquanto o fluxo AUTO CDC continua processando novos eventos.

Esse padrão unificado de preenchimento não suporta alvos SCD Tipo 2 ou bitemporais.

Exemplo: provisionar um alvo SCD durante uma migração

Um cenário comum de migração é uma tabela de dimensão que muda lentamente (SCD) que já existe em um sistema legado com anos de histórico acumulado, mas cujo feed original de mudanças não está mais disponível. Como os eventos de mudança originais desapareceram, você rejoga a história da tabela legada no novo alvo AUTO CDC uma vez, e então anexa um feed novo do CDC daqui para frente. Para mais informações sobre AUTO CDC e tipos de SCD, consulte As APIs AUTO CDC: simplifique a captura de dados de alterações com pipelines.

O padrão é um fluxo AUTO CDC executado uma única vez para a mesma tabela de streaming que o fluxo contínuo AUTO CDC tem como alvo. Um alvo AUTO CDC aceita apenas fluxos AUTO CDC, então a semente também deve ser um fluxo AUTO CDC. Um fluxo simples INSERT INTO ONCE de acréscimo para a mesma tabela falha na validação:

1. Crie a tabela de streaming de destino na qual seu fluxo AUTO CDC grava.
2. Inicialize o histórico legado uma vez com um AUTO CDC ONCE fluxo que lê a tabela SCD legada como um fluxo, sequenciado pela coluna legada de início da validade. Reprocesse as linhas legadas como eventos de alteração em vez de formatá-las manualmente. AUTO CDC cria as colunas de histórico __START_AT e __END_AT para um destino SCD Tipo 2, então não grave essas colunas diretamente.
3. Anexe o fluxo AUTO CDC em andamento que lê o feed de alterações mais recente. AUTO CDC resolve a ordem por chave, então a substituição deve ser válida para cada chave de negócio individualmente: a primeira mudança ativa de cada chave deve ser sequenciada após a última alteração propagada para essa mesma chave. Um valor de sequência que seja apenas posterior ao máximo global herdado ainda pode estar desatualizado para uma chave individual e, nesse caso, a primeira alteração efetiva dessa chave é ignorada ou ordenada incorretamente.

O código a seguir cria uma tabela de streaming que utiliza os passos acima:

CREATE OR REFRESH STREAMING TABLE customers_history;

-- One-time seed: replay the legacy history as change events
CREATE FLOW customers_history_seed
AS AUTO CDC ONCE INTO customers_history
FROM stream(legacy.customers_scd2)
KEYS (customer_id)
SEQUENCE BY valid_from
STORED AS SCD TYPE 2;

-- Ongoing live CDC into the same target
CREATE FLOW customers_history_cdc
AS AUTO CDC INTO customers_history
FROM stream(customers_cdc_bronze)
KEYS (customer_id)
SEQUENCE BY change_timestamp
STORED AS SCD TYPE 2;

Ambos os fluxos devem estar de acordo quanto às chaves, ao tipo SCD e ao tipo de dados da coluna de sequenciamento. No exemplo anterior, ambos os fluxos seguem uma sequência de carimbo de data/hora, que usa um único tempo de substituição para separar o histórico propagado do feed ativo. Se a tabela herdada for sequenciada por um valor de tipo diferente do feed ativo, converta um deles para que os tipos correspondam.

A mesma estrutura funciona para um alvo SCD Tipo 1: altere STORED AS SCD TYPE 2 para STORED AS SCD TYPE 1 em ambos os fluxos, e o alvo mantém apenas a linha atual para cada chave. Antes de confiar em qualquer uma das abordagens, valide, em uma amostra de chaves, se a primeira alteração em produção de uma chave propagada produz exatamente uma nova versão e encerra corretamente a versão anterior. Uma falha no sequenciamento para cada tecla geralmente aparece nessa etapa.

Recursos adicionais

• Carregar e processar dados incrementalmente com fluxos de pipeline do Lakeflow
• Usar fluxos em pipelines do Lakeflow
• As APIs AUTO CDC: simplificam a captura de dados de mudanças com pipelines
• append_flow
• create_auto_cdc_from_snapshot_flow
• create_auto_cdc_flow
• CREATE FLOW (pipelines)