Pular para o conteúdo principal
Última atualização em

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

info

Visualização

Esse recurso está em Prévia Pública.

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

Por default, um pipeline programará fluxos por suas dependências de dados: se um fluxo lê a tabela na qual outro fluxo grava, o leitor é executado após o gravador. A ordenação de fluxos permite que um fluxo aguarde outro do qual ele não lê, declarando explicitamente a dependência com depends_on.

depends_on="other_flow" significa que esse fluxo começa apenas após a conclusão bem-sucedida de other_flow. É uma aresta de agendamento: ela controla quando um fluxo começa, não como ele é executado. A declaração de depends_on não altera o trigger, o modo ou o fato de ser um fluxo único de um pipeline. Você ainda os declara no próprio fluxo, por exemplo, com once=True.

Quando usar a ordenação de fluxo​

O caso de uso principal é drenar data histórica antes de alternar para uma fonte ativa, preservando o estado de transmissão, como a migração de uma tabela de um preenchimento em lote para uma fonte ativa do Apache Kafka.

A leitura simultânea de ambas as fontes não funciona bem: o preenchimento retroativo limitado retarda a marca d'água, o que atrasa a remoção de estado e os resultados com janelas. O esvaziamento do preenchimento retroativo primeiro e, em seguida, a inicialização da transmissão ao vivo evitam isso. A ordenação de fluxos sequencia os dois.

A ordenação do fluxo também é compatível com outros padrões:

  • Ordenação rigorosa de vários preenchimentos na mesma tabela.
  • Pipelines em fases, como uma carga inicial, seguida de recuperação (catch-up) e, em seguida, um feed dinâmico.
  • Sequenciamento de fluxos que gravam em tabelas diferentes , mas devem ser executados em uma ordem definida.

Ordenar fluxos com depends_on​

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

O exemplo a seguir esvazia um preenchimento único na tabela de transmissão events e, em seguida, inicia uma transmissão ativa do Kafka na mesma tabela somente após a conclusão do preenchimento:

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

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

Python
@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 foreachBatch​

Utilize depends_on no fluxo de acréscimo que tem como alvo um coletor foreachBatch, não no decorador @dp.foreach_batch_sink. Um fluxo foreachBatch não pode ser um fluxo de uma única vez (once). Para saber mais sobre o comportamento do coletor e as considerações de nova tentativa, consulte Usar ForEachBatch para gravar em coletores de dados arbitrários em pipelines.

Fluxos foreachBatch de sequência em um pipeline Trigger​

Em um pipeline disparado, um fluxo foreachBatch pode ser um antecessor e um sucessor. O exemplo a seguir preenche uma tabela de transmissão, 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 lêem ordered_events. A dependência depends_on="export_events" sequencia o fluxo de auditoria após o fluxo de exportação, embora o fluxo de auditoria não leia a tabela de exportação. Substitua <catalog> e <schema> pelo catálogo e esquema das tabelas de origem e destino.

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

Execução de um fluxo foreachBatch após um backfill em um pipeline contínuo​

Em um pipeline contínuo, um fluxo foreachBatch pode ser um sucessor, mas não um antecessor. Seu predecessor deve ser um fluxo único (once). O exemplo a seguir processa as linhas preenchidas através de um sink foreachBatch após o término do preenchimento:

Python
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 uma única vez, portanto, o pipeline o rejeita como um predecessor com DEPENDS_ON_PREDECESSOR_MUST_BE_ONCE.

Requisitos e comportamento​

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

  • A ordenação de fluxos funciona apenas dentro de um pipeline. Sem o programador do pipeline, a ordenação não pode ser respeitada.
  • Um predecessor deve gravar em uma tabela ou coletor, não em uma view. Um fluxo que grava em uma view nunca atinge um estado terminal, de modo que um fluxo ordenado após ele nunca começaria. Os destinos da tabela de transmissão são predecessores válidos. Um fluxo de coletor foreachBatch pode ser um predecessor em um pipeline Trigger, mas não em um pipeline contínuo.
  • O pedido entre destinos é permitido. Um fluxo pode depender de um fluxo que grava em uma tabela diferente.
  • Nomes de fluxo desconhecidos e ciclos são detectados na validação , antes da execução do pipeline.

O que pode ser um predecessor em pipelines Trigger 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 em uma execução acionada atinge um estado terminal, de modo que a ordenação se aplica a cada execução.
  • Pipelines contínuos: um predecessor deve ser um fluxo de execução única (once) que atinge um estado terminal. Um fluxo que é executado continuamente nunca é encerrado, portanto, um fluxo ordenado após ele nunca começaria, e o pipeline o rejeita na validação.

Como um fluxo foreachBatch é sempre um destino de transmissão e não pode ser um fluxo de execução única, ele pode atuar como predecessor apenas em um pipeline com trigger. Em um pipeline contínuo, um fluxo foreachBatch pode aguardar um predecessor once, mas ele próprio não pode ser um predecessor.

Comportamento operacional​

O estado de conclusão de um predecessor once é durável, portanto, ele sobrevive a reinicializações e atualizações de pipeline:

  • Reiniciar: os fluxos já drenados permanecem drenados e são ignorados. O pipeline é retomado no primeiro fluxo que ainda não foi concluído. Depois que a execução atingir o fluxo dinâmico (live flow), as reinicializações posteriores retomam apenas o fluxo dinâmico.
  • Full refresh: limpa o estado de conclusão da cadeia e a executa novamente desde o início, em ordem.
  • Checkpoint Reset para um único fluxo: Reset o fluxo ativo reproduz apenas o fluxo ativo. Os preenchimentos retroativos once upstream permanecem esvaziados e não são executados novamente. Esta é a maneira usual de recuperar uma query ativa sem esvaziar a história novamente.

Recursos adicionais​

Nesta página