Ordenar la ejecución del flujo de la canalización con depends_on

Para usar depends_on, configura tu pipeline para usar el canal PREVIEW. Consulte channel en Configuraciones de canalización.

Por defecto, una tubería programa los flujos según sus dependencias de datos: si un flujo lee la tabla que otro escribe, el lector corre tras el escritor. El orden de los flujos permite que un flujo espere a otro del que no lee, declarando la dependencia explícitamente con depends_on.

depends_on="other_flow" significa que este flujo comienza solo después de other_flow completarse con éxito. Es una ventaja de planificación: controla cuándo empieza un flujo, no cómo se ejecuta. Declarar depends_on no cambia el desencadenante, el modo ni si es un flujo de un solo uso. Aún así declaras esos en el propio flujo, por ejemplo con once=True.

Cuándo usar el orden de flujo

El caso de uso principal consiste en vaciar los datos históricos antes de cambiar a una fuente en tiempo real, conservando al mismo tiempo el estado de la transmisión, como por ejemplo al migrar una tabla desde un rellenado por lotes a una fuente en tiempo real de Apache Kafka.

Leer ambas fuentes al mismo tiempo no funciona bien: el rellenado limitado frena el watermark, lo que retrasa la eliminación de estado y los resultados por ventanas. Vaciar primero el búfer y luego iniciar la transmisión en directo evita esto. La ordenación del flujo establece una secuencia entre ambos.

El orden de flujo también soporta otros patrones:

• Ordenar estrictamente varios rellenados en la misma tabla.
• Cadena de procesos por fases, como una carga inicial, seguida de una puesta al día y, a continuación, una alimentación en tiempo real.
• Flujos de secuenciación que escriben en diferentes tablas pero deben ejecutarse en un orden establecido.

Ordena los flujos con depends_on

depends_on está disponible en los decoradores @dp.append_flow y @dp.update_flow de la API de pipelines de Python. Acepta un único nombre de flujo o una lista de nombres de flujos. Con una lista, todo flujo nombrado debe completarse antes de que comience el flujo dependiente.

El siguiente ejemplo vacía un relleno único en la tabla de streaming events y, a continuación, inicia un flujo de Kafka en tiempo real en la misma tabla solo después de que finalice el relleno:

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 a más de un predecesor, hay que pasar una lista. El siguiente ejemplo ejecuta dos rellenos en paralelo y comienza la transmisión en directo solo después de que ambos se completen:

@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 los rellenos uno tras otro en lugar de en paralelo, encadena depends_on entre ellos:

@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 flujos foreachBatch

Use depends_on en el flujo de anexión que tiene como destino un receptor foreachBatch, no en el decorador @dp.foreach_batch_sink. Un foreachBatch flujo no puede ser un flujo único (once). Para consideraciones sobre comportamiento de sumideros y reintentos, véase Usar ForEachBatch para escribir en sumideros de datos arbitrarios en pipelines.

Secuenciar flujos foreachBatch en una canalización activada

En una canalización activada, un foreachBatch flujo puede ser tanto predecesor como sucesor. El siguiente ejemplo rellena una tabla de streaming, exporta sus filas mediante un sink foreachBatch y luego procesa las mismas filas mediante un segundo sink foreachBatch.

Ambos caudales de los sumideros muestran ordered_events. La depends_on="export_events" dependencia secuencia el flujo de auditoría después del flujo de exportación, aunque el flujo de auditoría no lea la tabla de exportación. Sustituye <catalog> y <schema> con el catálogo y el esquema de tus tablas de origen y 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")

Ejecutar un flujo foreachBatch después de un reabastecimiento en una canalización continua

En una pipeline continua, un foreachBatch flujo puede ser un sucesor, pero no un predecesor. Su predecesor debe ser un flujo único (once). El siguiente ejemplo procesa las filas de backfill a través de un foreachBatch sink después de que termina el backfill:

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

No añadas otro flujo con depends_on="export_events" a este ejemplo continuo. El flujo de exportación no es un flujo de ejecución única, por lo que el pipeline lo rechaza como predecesor con DEPENDS_ON_PREDECESSOR_MUST_BE_ONCE.

Requisitos y comportamiento

Las siguientes reglas se aplican al orden de flujo:

• El ordenamiento del flujo solo funciona dentro de una canalización. Sin el planificador de oleoductos, no se puede respetar el pedido.
• Un predecesor debe escribir en una tabla o un sumidero, no en una vista. Un flujo que escribe en una vista nunca alcanza un estado terminal, por lo que un flujo ordenado después de él nunca comenzaría. Los destinos de tabla de streaming son predecesores válidos. Un foreachBatch flujo de sumidero puede ser un predecesor en una canalización activada, pero no en una canalización continua.
• Se permite la realización de pedidos entre distintos destinos. Un flujo puede depender de un flujo que escribe en una tabla diferente.
• Los nombres de flujo no reconocidos y los ciclos se detectan durante la validación, antes de que se ejecute la canalización.

Qué puede ser un predecesor en canalizaciones desencadenadas y continuas

Los tipos de flujo que pueden actuar como predecesores dependen del modo de ejecución de la canalización:

• Pipelines desencadenados: cualquier flujo puede ser un predecesor. Cada flujo en una secuencia desencadenada alcanza un estado terminal, por lo que el orden se aplica dentro de cada ejecución.
• Tuberías continuas: un predecesor debe ser un flujo de un solo uso (once) que alcance un estado terminal. Un flujo que se ejecuta de forma continua nunca termina, por lo que un flujo ordenado después de él nunca arrancaría, y la tubería lo rechaza en la validación.

Dado que un flujo foreachBatch es siempre un sumidero de streaming y no puede ser un flujo único, solo puede actuar como predecesor en una canalización activada. En una tubería continua, un foreachBatch flujo puede esperar a un once predecesor, pero no puede ser un predecesor en sí mismo.

Comportamiento operativo

El estado de finalización de un predecesor once es persistente, por lo que se conserva tras los reinicios y las actualizaciones de la canalización:

• Reinicio: los flujos ya drenados permanecen drenados y se omiten. La canalización se reanuda a partir del primer flujo que aún no se ha completado. Una vez que la ejecución llega al flujo en vivo, los reinicios posteriores solo reanudan el flujo en vivo.
• Actualización completa: borra el estado de finalización de la cadena y la vuelve a ejecutar desde el principio, en orden.
• Restablecimiento del punto de control para un solo flujo: al restablecer el flujo activo, solo se vuelve a ejecutar el flujo activo. Los rellenos once de la fase anterior permanecen vaciados y no se vuelven a ejecutar. Esta es la forma habitual de recuperar una consulta en tiempo real sin volver a drenar el historial.

Recursos adicionales

• Cargue y procese datos de forma incremental con los flujos de canalización de Lakeflow
• Reposición de datos históricos con canalizaciones
• Usa ForEachBatch para escribir en sumideros de datos arbitrarios en pipelines
• append_flow
• update_flow