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

LakeFlow Pipelines における処理保証

Retries and reruns are inevitable in any real pipeline, so this page explains the processing guarantees Lakeflow pipelines give you and how to keep the parts you write safe to re-run.

Overview​

パイプラインの再実行が安全かどうかは、2つの関連するプロパティによって決まります：

  • べき等性 とは、同じ入力に対してパイプラインを何度実行しても、常に同じ結果が生成されることを意味します。失敗後の再実行、日付範囲のバックフィルの2回実行、またはジョブの手動再トリガーを行っても、重複する行が作成されたり、状態が破損したりすることはありません。
  • Processing guarantee describes how many times each record affects the result. At-least-once processing guarantees every record is processed, but a failure and retry might process some records more than once, which risks duplicates. Exactly-once processing guarantees every record affects the result as if it were processed precisely one time, even across retries, with no duplicates and no gaps.

Lakeflow pipelines are idempotent by default for the pieces they manage, and give you exactly-once processing within their own managed tables. The important thing to understand is where those guarantees stop being automatic, so you can add the right safeguards at the edges of your pipeline.

仕組み​

Lakeflow Pipelines は、管理するフローに対して Exactly-Once 処理と冪等性を提供し、記述するロジックも冪等に保つためのツールを提供します。

マネージドテーブルのExactly-once処理​

マネージドテーブル内では、defaultで exactly-once（一度だけ）処理が提供されます。ストリーミングテーブルは、Structured StreamingのチェックポイントとDelta Lakeのトランザクション書き込みを組み合わせて使用します。各マイクロバッチはソースのオフセットと出力をまとめてcommitするため、障害後に再試行されたバッチは完全に成功するか、完全にロールバックされて再試行されるかのいずれかとなり、部分的に2回適用されることはありません。これは、Auto Loaderによるファイルの取り込み、Kafka、Kinesis、Azure Event Hubsの読み取り、および AUTO CDC のアップサートに対して、ユーザー側でコードを追加することなく適用されます。

at-least-once（少なくとも1回）ソースが同じレコードを複数回送信した場合、パイプラインはそれらを一意のレコードとして処理し、すべてをテーブルに書き込みます。それらの重複を削除することは、お客様の責任となります。at-least-onceソースの重複を削除するをご覧ください。

読み取りのべき等性も、それらの同じチェックポイントから導かれます。Auto Loader とストリーミングテーブルのチェックポイントにより、状態追跡の目的で各ソースファイルまたはオフセットが確実に1回処理されます。そのため、障害後にパイプラインの更新を再処理する場合、データの再処理やスキップを行うのではなく、チェックポイントから再開されます。これは、手動で作成したバッチループの代わりに spark.readStream でストリーミングテーブルを使用することで実現できます。See ストリーミングテーブル.

手動記述の MERGE の代わりに AUTO CDC を使用する​

AUTO CDC INTO は、その keys および sequence_by に関して本質的にべき等です。同じ変更レコードを2回適用したり、レコードを順序不同で適用したりしても、最終的な状態は同じになります。これは、パイプラインがシーケンス列を使用して、受信した行が格納されている行よりも実際に新しいかどうかを判断するためです：

SQL
CREATE FLOW customers_cdc_flow AS AUTO CDC INTO customers_silver
FROM stream(customers_cdc_bronze)
KEYS (customer_id)
SEQUENCE BY sequence_num
STORED AS SCD TYPE 1;

If you write your own upsert logic outside of AUTO CDC (rare, but sometimes necessary for complex merge conditions), key it on a stable business key and make it safe to apply twice, for example a MERGE ... WHEN MATCHED keyed on order_id rather than a blind INSERT. For more information, see The AUTO CDC APIs: Simplify change data capture with pipelines.

独自の変換をべき等に保つ​

書き込み操作を再実行する際にロジックをべき等に保つには、次の2つのガイドラインに従ってください：

  • Avoid non-deterministic transformations in materialized views. Because a materialized view can fully or incrementally recompute, avoid functions whose output depends on when they run rather than what the input is. For example, don't use current_timestamp() to compute a business value that should stay fixed once written; take the timestamp from the source event or pass it in as a parameter so recomputation produces identical output.
  • フル更新が安全に行われるように設計してください。 フル更新を行うとテーブルが削除され、最初から再計算されます。これは、すべてのアップストリームソースが完全な履歴を生成できる場合にのみ安全です。アップストリームソースが変更のローリングウィンドウのみを公開している場合、ダウンストリームの AUTO CDC テーブルをフルリフレッシュすると履歴が警告なしに失われる可能性があるため、この点を考慮してソースとトピックの保持を設計してください。

エッジで「厳密に 1 回」の処理を実現する​

Exactly-once（厳密に1回）が自動でなくなるのは、外部システムへの書き込みなど、パイプラインが直接制御する範囲の境界においてです。外部システムにファンアウトする場合は、書き込み自体をべき等にするようにしてください。たとえば、受信側でキーによるアップサートを行うなどの方法があります。これを行わないと、再試行されたマイクロバッチによって同じバッチが二重に書き込まれる可能性があるためです。以下のシンクは、エグゼキューターからバッチの各パーティションを書き込み、べき等性キーを使用して、再試行されたバッチが二重書き込みされないようにします。

Python
from pyspark import pipelines as dp

@dp.foreach_batch_sink(name="orders_to_external_api")
def write_orders_to_api(batch_df, batch_id):
    def write_partition(rows):
        # Open one client per partition.
        for row in rows:
            # Use an idempotency key (order_id) so a retried batch doesn't double-write.
            upsert_to_external_system(key=row.order_id, payload=row.asDict())

    batch_df.select("order_id", "amount").foreachPartition(write_partition)

外部システムへの書き込みの詳細については、「LakeFlow Pipelines」を参照してください。

Deduplicate at-least-once sources​

ソースがレコードを複数回配信する可能性がある場合は、ダウンストリームで重複排除を行います。ウォーターマークを dropDuplicatesWithinWatermark と組み合わせます。これはウォーターマークを認識し、重複を検出するために無制限の状態を必要としません。イベントを一意に識別する列で重複排除を行います。単一の列だけでは一意にならない場合、ID は複数の列にまたがることができます。次の例では、クリックシーケンス番号はセッション内でのみ一意であるため、これら 2 つの列を組み合わせてイベントを識別します。

Python
from pyspark import pipelines as dp

@dp.table(name="clicks_deduped")
def clicks_deduped():
    return (
        spark.readStream.table("clicks_bronze")
        .withWatermark("click_ts", "5 minutes")
        .dropDuplicatesWithinWatermark(["session_id", "click_seq_num"])
    )

Choose those columns from the source's uniqueness contract, not from what looks distinct in sample data. Columns that can legitimately repeat discard real events when you treat them as the identity. A user clicking the same ad twice is a common example: deduplicating on the user and the ad silently drops the second click.

AUTO CDCのキーベースのアップサートセマンティクスも重複を自然に解消するため、安定したビジネスキーでキー設定された AUTO CDC フローを介して「最低 1 回」のデータをルーティングすることも、「厳密に 1 回」の状態に収束させる別の方法です。

制限事項​

「正確に1回」の処理は、マネージド型 Delta-to-Delta フローに適用されます。以下のエッジを「少なくとも1回」として扱い、明示的な重複排除またはべき等書き込みロジックを追加してください：

  • foreach_batch_sink およびカスタム外部書き込み。 Spark はバッチが少なくとも 1 回は 試行 されることを保証しますが、部分的な書き込みの後に再試行されたバッチでは、一部の行が外部システムで 2 回表示される可能性があります。ナチュラルキーでのアップサートや、受信側で重複排除可能なバッチIDの書き込みなどを行い、外部書き込みをべき等にします。
  • Kafka as a sink. Kafka topics don't support transactional exactly-once writes the way Delta does, so a retried micro-batch writing to Kafka can produce duplicate messages. If downstream consumers are sensitive to duplicates, dedupe on the consumer side, for example by event ID.
  • Custom Python data sources used as sources. Whether reads are exactly-once depends on whether your source implementation correctly reports and resumes from offsets. If it doesn't track offsets, treat it as at-least-once and dedupe downstream with dropDuplicates on an event ID or by relying on AUTO CDC's key-based upsert semantics.

経験則として、パイプライン全体が Delta-to-Delta（マネージドフローを通じて Delta テーブルを読み書きするストリーミングテーブルおよびマテリアライズドビュー）である場合、すでに exactly-once（厳密に1回）が実現されています。foreach_batch_sink、非 Delta シンク、または未検証のカスタムソースを追加した時点で、その特定の境界を at-least-once（少なくとも1回）として扱い、そこにべき等な書き込みまたは重複排除のロジックを追加してください。

その他のリソース​

このページの見出し