depends_on による注文パイプラインフロー実行

Important

depends_onを使うには、パイプラインをプレビューチャネルを使うように設定してください。 channelのを参照してください。

デフォルトでは、パイプラインはデータの依存関係によってフローをスケジューリングします。あるフローが別のフローが書き込むテーブルを読み取ると、リーダーはライターの後に動きます。フローの順序付けでは、depends_on を使用して依存関係を明示的に宣言することで、あるフローが、そこからデータを読み取らない別のフローを待機させることができます。

depends_on="other_flow" このフローは other_flow が成功裏に完了した後に始まることを意味します。これはスケジューリングのエッジであり、フローの開始 時期を制御 するものであり、 どのように実行するか を決めるものではありません。 depends_onを宣言しても、フローのトリガーやモード、または一度きりのフローかどうかは変わりません。例えば once=Trueでは、フロー自体でそれらを宣言します。

フロー順序付けの使用タイミング

主なユースケースは、ストリーミング状態を保持しながらライブソースに切り替える前に過去データを抽出することです。例えば、バッチバックフィルからライブApache Kafkaソースへの移行などです。

両方のソースを同時に読み取るのはうまく機能しません。有界バックフィルがウォーターマークの進行を妨げるため、ステートの削除とウィンドウ結果の出力が遅れます。まず埋め戻しの水を抜き、その後ライブ配信を始めることでこの問題を回避できます。フローの順序付けはこの二つを順序付けます。

フローの順序付けは他のパターンもサポートしています:

同じテーブルに対する複数のバックフィルを厳密な順序で実行する。
フェーズ分けされたパイプライン、例えば初期負荷、その後キャッチアップ、ライブフィードなどです。
異なるテーブルに書き込みますが、決まった順序で実行しなければならないシーケンスフローです。

depends_on を用いた順序制御フロー

depends_onPythonパイプラインAPIの@dp.append_flowおよび@dp.update_flowデコレーターで利用可能です。単一のフロー名または複数のフロー名のリストを受け付けます。リストの場合、すべての名前付きフローが依存フローが始まる前に完了しなければなりません。

以下の例では、1 回限りのバックフィルを events ストリーミングテーブルに流し込み、そのバックフィルが完了した後でのみ、同じテーブルへのライブ Kafka ストリームを開始します。

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

複数の先行処理を待機するには、リストを渡します。以下の例は2つのバックフィルを並行して実行し、両方が完了した後にのみライブストリームを開始します。

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

バックフィルを並列ではなく順番に実行するには、それらの間で 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")

foreachBatch フローの順序

depends_onはforeachBatchのシンクをターゲットにした付加フローに使い、@dp.foreach_batch_sinkデコレーターには使いません。 foreachBatchフローは一度きり(once)フローにはなりません。シンクの挙動や再試行の考慮事項については、「パイプライン内の任意のデータシンクに書き込みするためのUse ForEachBatch」を参照してください。

トリガーされたパイプライン内の foreachBatch フローのシーケンス

トリガー付きパイプラインでは、foreachBatchフローは先行フローであると同時に後続フローでもあり得ます。以下の例はストリーミングテーブルをバックフィルし、その行を foreachBatch シンクにエクスポートし、同じ行を2つ目の foreachBatch シンクで処理します。

両方のシンクフローは ordered_events を示します。 depends_on="export_events"依存関係は、監査フローがエクスポートテーブルを読み取っなくても、エクスポートフローの後に監査フローをシーケンスします。 <catalog> と <schema> を、それぞれソーステーブルと宛先テーブルのカタログおよびスキーマに置き換えてください。

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

連続パイプラインのバックフィル後にforeachBatchフローを実行します

連続パイプラインでは、foreachBatchフローは後続フローにはなり得ますが、先行フローにはなりません。その前身は一度きり(once)フローでなければなりません。以下の例は、埋め戻しが終わった後に foreachBatch のシンクで埋め戻された行を処理しています:

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

この連続例に depends_on="export_events" を含む別のフローを追加しないでください。エクスポートフローは一度きりのフローではないため、パイプラインは DEPENDS_ON_PREDECESSOR_MUST_BE_ONCEで先行フローとして拒否します。

要件と動作

フローの順序付けには以下のルールが適用されます:

フローの順序付けはパイプライン内でのみ機能します。 パイプラインスケジューラがなければ、注文は尊重できません。
先行処理プログラムは、ビューではなく、テーブルまたはシンクに書き込む必要があります。 ビューに書き込みを行うフローは終末状態に到達せず、その後に順序付けられたフローは決して開始されません。ストリーミングテーブルターゲットは有効な先行対象です。 foreachBatchシンクフローはトリガーされたパイプラインの前駆体となり得ますが、連続パイプラインでは前駆体にはなれません。
クロスデスティネーション注文も許可されています。 フローは別のテーブルに書き込むフローに依存します。
未知のフロー名やサイクルは、パイプライン実行前の検証時に検出されます。

トリガー付きおよび連続的パイプラインの前身となりうるもの

前段階として機能できるフローの種類は、パイプラインの実行モードによって異なります。

トリガー付きパイプライン: 任意のフローが前段階となり得ます。トリガーされた実行のすべてのフローは終端状態に達するため、順序は各実行内で適用されます。
連続パイプライン: 前段階は終端状態に達する一度きり(once)フローでなければなりません。連続的に実行されるフローは終了しないため、その後に順序付けられたフローは開始されず、検証時にパイプラインはそれを拒否します。

foreachBatchフローは常にストリーミングシンクであり、一度きりのフローではないため、トリガーされたパイプライン内でのみ前段階として機能します。連続パイプラインでは、 foreachBatch フローは once 前段階を待つことはできますが、それ自体が前段階であることはできません。

動作挙動

onceの前任者の完了状態は耐久性があるため、再起動やパイプラインの更新にも耐えられます:

再起動: すでにドレイン済みのフローは、そのままドレイン済みとして扱われ、スキップされます。パイプラインはまだ完了していない最初の流れから再開されます。実行がライブフローに到達した後、その後の再起動はライブフローのみを再開します。
フルリフレッシュ: チェーンの完了状態をクリアし、順番通りに最初から再実行します。
単一のフローに対するチェックポイントリセット: ライブフローをリセットすると、ライブフローのみが再生されます。上流の once バックフィルは処理し尽くされると、再実行されません。これは、履歴を再ドレインせずにライブクエリを復元する一般的な方法です。

その他のリソース

フィードバック

このページはお役に立ちましたか?

最終更新: 2026-10-16