Ordenar a execução do fluxo do pipeline com depends_on

Para usar depends_on, configure o seu pipeline para usar o canal PREVIEW. Veja channel em Configurações do oleoduto.

Por defeito, um pipeline agenda a execução dos fluxos de acordo com as respetivas dependências de dados: se um fluxo lê a tabela que outro fluxo escreve, o primeiro é executado depois do segundo. A ordenação entre fluxos permite que um fluxo espere por outro fluxo do qual não faz leitura, declarando explicitamente a dependência com depends_on.

depends_on="other_flow" significa que este fluxo só começa depois de other_flow ser concluído com sucesso. É uma vantagem de agendamento: controla quando um fluxo começa, não como ele corre. Declarar depends_on não altera o gatilho, modo ou se é um fluxo único. Ainda declaras esses elementos no próprio fluxo, por exemplo com once=True.

Quando usar a ordenação de fluxo

O principal caso de uso é drenar dados históricos antes de mudar para uma fonte em direto, preservando o estado de streaming, como migrar uma tabela de um backfill batch para uma fonte Apache Kafka em direto.

Ler ambas as fontes ao mesmo tempo não funciona bem: o retropreenchimento limitado atrasa o watermark, o que atrasa a eliminação do estado e os resultados das janelas. Esvaziar primeiro o backfill e só depois iniciar a transmissão em direto evita isto. A ordenação do fluxo coloca os dois em sequência.

A ordenação de fluxo também suporta outros padrões:

• Ordenação estrita de vários backfills para a mesma tabela.
• Pipelines faseados, como uma carga inicial, depois uma recuperação, e depois uma transmissão em tempo real.
• Fluxos de sequenciação que escrevem em diferentes tabelas mas têm de ser executados numa ordem definida.

Fluxos de encomendas com depends_on

depends_on está disponível nos decoradores @dp.append_flow e @dp.update_flow na API de pipelines em Python. Aceita um único nome de fluxo ou uma lista de nomes de fluxos. Com uma lista, cada fluxo nomeado deve ser concluído antes que o fluxo dependente comece.

O exemplo seguinte carrega uma operação única de backfill para a tabela de streaming events, e depois inicia um streaming Kafka em tempo real na mesma tabela só depois de o backfill terminar:

from pyspark import pipelines as dp

dp.create_streaming_table(name="events")

# Drain the historical backfill first.
@dp.append_flow(target="events", once=True, name="events_backfill")
def events_backfill():
    return spark.read.table("historical_events")

# Start the live stream only after the backfill completes.
@dp.append_flow(target="events", name="events_live", depends_on="events_backfill")
def events_live():
    return (
        spark.readStream.format("kafka")
        .option("kafka.bootstrap.servers", "<server>:<port>")
        .option("subscribe", "events")
        .load()
    )

Para esperar por mais do que um antecessor, passa uma lista. O exemplo seguinte executa dois backfills em paralelo e inicia o fluxo em tempo real apenas depois de ambos estarem concluídos:

@dp.append_flow(target="events", once=True, name="backfill_2024")
def backfill_2024():
    return spark.read.table("events_2024")

@dp.append_flow(target="events", once=True, name="backfill_2025")
def backfill_2025():
    return spark.read.table("events_2025")

@dp.append_flow(
    target="events",
    name="events_live",
    depends_on=["backfill_2024", "backfill_2025"],
)
def events_live():
    return spark.readStream.format("kafka").option("subscribe", "events").load()

Para executar os preenchimentos retroativos de forma sequencial, em vez de em paralelo, encadeie-os com depends_on:

@dp.append_flow(target="events", once=True, name="backfill_2024")
def backfill_2024():
    return spark.read.table("events_2024")

@dp.append_flow(
    target="events", once=True, name="backfill_2025", depends_on="backfill_2024"
)
def backfill_2025():
    return spark.read.table("events_2025")

Ordem dos fluxos foreachBatch

Use depends_on no fluxo append que aponta para o foreachBatch sink, não para o @dp.foreach_batch_sink decorator. Um foreachBatch fluxo não pode ser um fluxo único (once). Para considerações sobre o comportamento dos destinos de dados e as retentativas, veja Utilizar o ForEachBatch para escrever em destinos de dados arbitrários em pipelines.

Sequência de fluxos foreachBatch num pipeline desencadeado

Num pipeline desencadeado, um foreachBatch fluxo pode ser tanto um antecessor como um sucessor. O exemplo seguinte preenche uma tabela de streaming, exporta as suas linhas através de um foreachBatch sink e depois processa as mesmas linhas através de um segundo foreachBatch sink.

Ambos os fluxos do sink indicam ordered_events. A dependência depends_on="export_events" sequencia o fluxo de auditoria após o fluxo de exportação, mesmo que o fluxo de auditoria não leia a tabela de exportação. Substitua <catalog> e <schema> com o catálogo e o esquema das suas tabelas de origem e destino.

import dlt
from pyspark import pipelines as dp

dp.create_streaming_table(name="ordered_events")


@dp.append_flow(target="ordered_events", once=True, name="events_backfill")
def events_backfill():
    return spark.read.table("<catalog>.<schema>.historical_events")


@dp.foreach_batch_sink(name="events_export_sink")
def export_batch(batch_df, batch_id):
    batch_df.write.format("delta").mode("append").saveAsTable(
        "<catalog>.<schema>.exported_events"
    )


@dp.append_flow(
    target="events_export_sink", name="export_events", depends_on="events_backfill"
)
def export_events():
    return dlt.read_stream("ordered_events")


@dp.foreach_batch_sink(name="events_audit_sink")
def audit_batch(batch_df, batch_id):
    batch_df.write.format("delta").mode("append").saveAsTable(
        "<catalog>.<schema>.audited_events"
    )


@dp.append_flow(
    target="events_audit_sink", name="audit_events", depends_on="export_events"
)
def audit_events():
    return dlt.read_stream("ordered_events")

Execute um fluxo foreachBatch após um backfill numa pipeline contínua

Num pipeline contínuo, um foreachBatch fluxo pode ser um sucessor, mas não um antecessor. O seu antecessor deve ser um fluxo de utilização única (once). O exemplo seguinte processa as linhas preenchidas retroativamente através de um foreachBatch sink após o final do preenchimento:

import dlt
from pyspark import pipelines as dp

dp.create_streaming_table(name="backfilled_events")


@dp.append_flow(target="backfilled_events", once=True, name="events_backfill")
def events_backfill():
    return spark.read.table("<catalog>.<schema>.historical_events")


@dp.foreach_batch_sink(name="events_export_sink")
def export_batch(batch_df, batch_id):
    batch_df.write.format("delta").mode("append").saveAsTable(
        "<catalog>.<schema>.continuous_exported_events"
    )


@dp.append_flow(
    target="events_export_sink", name="export_events", depends_on="events_backfill"
)
def export_events():
    return dlt.read_stream("backfilled_events")

Não adicione outro fluxo com depends_on="export_events" a este exemplo contínuo. O fluxo de exportação não é um fluxo de execução única, pelo que o pipeline rejeita-o como antecessor com DEPENDS_ON_PREDECESSOR_MUST_BE_ONCE.

Requisitos e comportamento

As seguintes regras aplicam-se à ordenação de fluxos:

• A ordenação do fluxo funciona apenas dentro de uma canalização. Sem o agendador de pipeline, a encomenda não pode ser cumprida.
• Um antecessor deve escrever para uma tabela ou piso, não para uma vista. Um fluxo que escreve numa visualização nunca atinge um estado terminal, por isso um fluxo definido após esse nunca se iniciaria. Os alvos de tabela de streaming são predecessores válidos. Um foreachBatch fluxo sink pode ser um predecessor num pipeline acionado, mas não num pipeline contínuo.
• É permitido fazer encomendas para vários destinos. Um fluxo pode depender de um fluxo que escreve para uma tabela diferente.
• Nomes e ciclos de fluxo desconhecidos são detetados na validação, antes de o pipeline funcionar.

O que pode ser um precedente em pipelines acionados e contínuos

Os tipos de fluxo que podem atuar como predecessores dependem do modo de execução do pipeline:

• Pipelines acionados: qualquer fluxo pode ser um predecessor. Cada fluxo numa execução desencadeada atinge um estado terminal, pelo que a ordem aplica-se em cada execução.
• Pipelines contínuos: um antecessor deve ser um fluxo único (once) que atinge um estado terminal. Um fluxo que é executado continuamente nunca termina, por isso um fluxo configurado a seguir nunca seria iniciado, e o pipeline rejeita-o durante a validação.

Como foreachBatch é sempre um sink de transmissão e não pode ser um fluxo de execução única, só pode atuar como antecessor num pipeline acionado por gatilho. Numa pipeline contínua, um fluxo foreachBatch pode depender de um antecessor once, mas esse próprio fluxo não pode ser um antecessor.

Comportamento operacional

O estado de conclusão de um predecessor once é persistente, pelo que sobrevive a reinícios e atualizações do pipeline:

• Reiniciar: os fluxos já drenados permanecem drenados e são ignorados. O oleoduto retoma no primeiro fluxo que ainda não foi concluído. Quando a execução atinge o fluxo em direto, os reinícios subsequentes retomam apenas o fluxo em direto.
• Atualização completa: limpa o estado de conclusão da cadeia e volta a executá-la do início, por ordem.
• Resetar o checkpoint para um único fluxo: reiniciar o fluxo ao vivo reproduz apenas o fluxo ao vivo. Os preenchimentos a montante once mantêm-se drenados e não são reutilizados. Esta é a forma habitual de recuperar uma consulta em tempo real sem voltar a drenar o histórico.

Recursos adicionais

• Carregar e processar dados de forma incremental com os fluxos do oleoduto Lakeflow
• Preenchimento de dados históricos com pipelines
• Use ForEachBatch para escrever em sumidouros de dados arbitrários em pipelines
• append_flow
• update_flow