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

Para usar depends_on, configure seu pipeline para usar o canal PREVIEW. Consulte channel nas configurações de Pipeline.

Por padrão, um pipeline programa os fluxos com base em suas dependências de dados: se um fluxo lê a tabela que outro fluxo grava, o fluxo leitor é executado depois do fluxo gravador. O controle de fluxo permite que um fluxo espere por outro fluxo do qual ele não lê, declarando a dependência explicitamente com depends_on.

depends_on="other_flow" significa que esse fluxo só começa após other_flow ser concluído com sucesso. É uma vantagem de agendamento: controla quando um fluxo começa, não como ele roda. Declarar depends_on não altera o gatilho, o modo ou se ele é um fluxo único. Você ainda declara esses 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 alternar para uma fonte em tempo real, preservando o estado do fluxo, como migrar uma tabela de um preenchimento em lote para uma fonte Apache Kafka em tempo real.

Ler ambas as fontes ao mesmo tempo não funciona bem: o preenchimento limitado retém a marca d'água, o que atrasa a remoção do estado e os resultados em janelas. Drenar o preenchimento primeiro e, em seguida, iniciar o fluxo em tempo real evita isso. A ordem de fluxo sequencia os dois.

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

• Ordenar estritamente vários preenchimentos na mesma tabela.
• Pipelines em fases, como uma carga inicial, seguida de uma recuperação e, em seguida, um fluxo em tempo real.
• Fluxos de sequenciamento que escrevem em tabelas diferentes , mas precisam rodar em uma ordem definida.

Ordene os fluxos 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, todo fluxo nomeado deve ser concluído antes que o fluxo dependente comece.

O exemplo a seguir carrega um backfill único na tabela de streaming events e então inicia um fluxo do Kafka em tempo real na mesma tabela somente após a conclusão do backfill:

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 de um antecessor, passe uma lista. O exemplo a seguir executa dois backfills em paralelo e inicia a transmissão ao vivo somente após a conclusão de ambos:

@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 ordenar os preenchimentos um após o outro, em vez de em paralelo, encadeie depends_on entre eles:

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

Ordenar fluxos do foreachBatch

Use depends_on no fluxo de acréscimo que tem como destino um coletor foreachBatch, e não no decorador @dp.foreach_batch_sink. Um fluxo foreachBatch não pode ser um fluxo de execução única (once). Para informações sobre o comportamento de destinos de dados e considerações sobre novas tentativas, consulte Use o ForEachBatch para gravar em destinos de dados arbitrários em pipelines.

Fluxos foreachBatch em um pipeline acionado

Em um pipeline acionado, um foreachBatch fluxo pode ser tanto um predecessor quanto um sucessor. O exemplo a seguir preenche retroativamente uma tabela de streaming, exporta suas linhas por meio de um coletor foreachBatch e, em seguida, processa as mesmas linhas por meio de um segundo coletor foreachBatch.

Ambos os fluxos de sink mostram 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> pelo catálogo e 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 em um pipeline contínuo

Em um pipeline contínuo, um foreachBatch fluxo pode ser um sucessor, mas não um predecessor. Seu predecessor deve ser um fluxo de ocorrência única (once). O exemplo a seguir processa as linhas recuperadas por meio de um coletor foreachBatch após a conclusão da recuperação de dados:

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" neste exemplo contínuo. O fluxo de exportação não é um fluxo de execução única; por isso, o pipeline o rejeita como predecessor, indicando DEPENDS_ON_PREDECESSOR_MUST_BE_ONCE.

Requisitos e comportamento

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

• A ordenação de fluxo funciona apenas dentro de um pipeline. Sem o agendador de pipeline, a ordenação não pode ser respeitada.
• Um predecessor deve escrever em uma tabela ou coletor, não em uma visão. Um fluxo que escreve em uma visualização nunca atinge um estado terminal, então um fluxo ordenado depois dele nunca começaria. Alvos de tabelas de streaming são predecessores válidos. Um foreachBatch fluxo de sumidouro pode ser um predecessor em um pipeline acionado, mas não em um pipeline contínuo.
• Ordens entre destinos são permitidas. Um fluxo pode depender de um fluxo que grava em uma tabela diferente.
• Nomes e ciclos de fluxo desconhecidos são capturados na validação, antes que o pipeline seja executado.

O que pode ser um predecessor 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. Todo fluxo em uma execução acionada atinge um estado terminal, então a ordem se aplica dentro de cada execução.
• Pipelines contínuos: um predecessor deve ser um fluxo único (once) que atinge um estado terminal. Um fluxo que é executado continuamente nunca termina, então um fluxo definido após ele nunca começaria, e o pipeline o rejeita durante a validação.

Como um fluxo foreachBatch é sempre um coletor de streaming e não pode ser um fluxo único, ele só pode atuar como predecessor em um pipeline acionado. Em um pipeline contínuo, um fluxo foreachBatch pode aguardar um antecessor once, mas ele próprio não pode ser um antecessor.

Comportamento operacional

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

• Reiniciar: fluxos já drenados permanecem drenados e são ignorados. O gasoduto retoma no primeiro fluxo que ainda não foi concluído. Após a execução atingir o fluxo ao vivo, reinicializações posteriores retomam apenas o fluxo ao vivo.
• Atualização completa: limpa o estado de conclusão da cadeia e a executa novamente desde o início, em ordem.
• Redefinição do checkpoint para um único fluxo: redefinir o fluxo ativo reexecuta apenas o fluxo ativo. Os preenchimentos once upstream permanecem drenados e não são executados novamente. Essa é a forma usual de recuperar uma consulta ao vivo sem drenar novamente o histórico.

Recursos adicionais

• Carregar e processar dados incrementalmente com fluxos de pipeline do Lakeflow
• Reabastecimento de dados históricos usando pipelines
• Use o ForEachBatch para gravar em destinos de dados arbitrários em pipelines
• append_flow
• update_flow