メインコンテンツまでスキップ
に最終更新

depends_on を使用したパイプライン フローの実行順序付け

備考

プレビュー

この機能は パブリック プレビュー段階です。

depends_on を使用するには、LakeFlow Pipelines PREVIEW チャンネルを使用するようにパイプラインを構成します。「パイプライン構成」の channel を参照してください。

default では、パイプラインはデータの依存関係に基づいてフローをスケジュールします。あるフローが別のフローが書き込むテーブルを読み取る場合、読み取り側は書き込み側の後に実行されます。フローの順序付けにより、depends_on を使用して依存関係を明示的に宣言することで、フローが読み取りを行って いない 別のフローを待機できるようになります。

depends_on="other_flow" other_flow が正常に完了した後にのみ、このフローが起動することを意味します。これはスケジュール上のエッジであり、フローが どのように ランされるかではなく、 いつ 起動されるかを制御します。depends_onを宣言しても、フローのTrigger、モード、または1回限りのフローであるかどうかは変わりません。それらはフロー自体で宣言します（例：once=Trueを使用）。

フローの順序付けを使用するタイミング​

主なユースケースは、ストリーミング状態を維持しながらライブソースに切り替える前にヒストリカルデータをドレインすることです (バッチバックフィルからライブの Apache Kafka ソースへのテーブルの移行など)。

両方のソースを同時に読み取ることはうまく機能しません。バインドされたバックフィルによってウォーターマークが保持され、ステートの削除やウィンドウ処理の結果が遅延するためです。最初にバックフィルをドレインし、その後にライブストリームを開始することで、これを回避できます。フローの順序付けによって、これら2つが順序付けられます。

フローの順序付けでは、他のパターンもサポートされています。

  • 複数のバックフィルを同じテーブルに厳密に順序付けして実行します。
  • 初期ロード、キャッチアップ、ライブフィードなどの段階的なパイプライン。
  • 異なる テーブルに書き込みを行い、特定の順序で実行する必要があるフローの順序付け。

depends_on を使用したフローの順序付け​

depends_on は、Pythonパイプライン API の @dp.append_flow および @dp.update_flow デコレーターで使用できます。単一のフロー名またはフロー名のリストを受け入れます。リストを使用する場合、依存するフローが開始される前に、名前が付けられたすべてのフローが完了する必要があります。

次の例では、1回限りのバックフィルを events ストリーミングテーブルにドレインし、バックフィールの完了後にのみ、同じテーブルへのライブ Kafka ストリームを起動します。

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

複数の前提条件を待機させるには、リストを渡します。次の例では、2つのバックフィルを並列実行し、両方が完了した後にのみライブストリームを開始します。

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

バックフィルを並列ではなく順番に実行するには、それらの間で depends_on をチェーンします。

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

要件と動作​

次のルールは、フローの順序付けに適用されます。

  • フローの順序付けは、パイプライン内でのみ機能します。 パイプライン スケジューラがない場合、順序付けを適用できません。
  • 前提条件となる処理は、ビューではなく、テーブルまたはシンクに書き込む必要があります。 ビューに書き込むフローは終了状態に到達しないため、その後に順序付けられたフローが起動することはありません。ストリーミングテーブルのターゲットと foreachBatch のシンクは、どちらも有効な先行処理です。
  • クロスデスティネーションの順序付けが許可されます。 フローは、別のテーブルに書き込むフローに依存できます。
  • 不明なフロー名と循環参照は、パイプラインのランの前に、検証時に検出されます 。

Triggerパイプラインおよび継続的パイプラインの前身となる要素​

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

  • Triggered pipelines: 任意のフローを前提条件とすることができます。Triggerラン内のすべてのフローは終了状態に到達するため、順序付けは各ラン内で適用されます。
  • 連続パイプライン： 先行フローは、ターミナル状態に達する1回限りの (once) フローである必要があります。連続実行されるフローは終了しないため、その後に順序付けられたフローは起動せず、パイプラインは検証時にそれを拒否します。

foreachBatch フローは常にストリーミング シンクであり、1回限りのフローにはできないため、Trigger パイプラインでのみ先行フローとして機能できます。連続パイプラインでは、foreachBatch フローは once 先行フローを待機できますが、それ自体が先行フローになることはできません。

動作​

once の前提条件の完了状態は永続的であるため、再起動やパイプラインの更新後も保持されます。

  • 再起動: すでにドレインされているフローはドレイン状態のまま維持され、スキップされます。パイプラインは、まだ完了していない最初のフローから再開されます。実行がライブ フローに到達した後は、その後の再起動ではライブ フローのみが再開されます。
  • 完全更新： チェーンの完了状態をクリアし、最初から順番に再実行します。
  • 単一フローのチェックポイントのReset： ライブフローのResetでは、ライブフローのみが再再生されます。上流の once バックフィルはドレインされたまま保持され、再実行されません。これは、履歴を再ドレインせずにライブクエリーを復旧する通常の方法です。

その他のリソース​

このページの見出し