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

LakeFlow Pipelines における処理保証

どのような実際のパイプラインにおいても再試行や再実行は避けられないため、このページでは、LakeFlow Pipelinesが提供する処理保証と、作成したコードを安全に再実行できるようにする方法について説明します。

概要​

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

  • べき等性 とは、同じ入力に対してパイプラインを何度実行しても、常に同じ結果が生成されることを意味します。失敗後の再実行、日付範囲のバックフィルの2回実行、またはジョブの手動再トリガーを行っても、重複する行が作成されたり、状態が破損したりすることはありません。
  • 処理保証 は、各レコードが結果に何回影響を与えるかを示します。 At-least-once （少なくとも 1 回）の処理では、すべてのレコードが処理されることが保証されますが、障害や再試行によって一部のレコードが複数回処理される可能性があり、重複のリスクがあります。 Exactly-once （1 回限り）の処理では、再試行が発生した場合でも、重複や欠落がなく、すべてのレコードが正確に 1 回処理されたかのように結果に反映されることが保証されます。

Lakeflow pipelines は、管理対象の要素に対して default でべき等であり、独自のマネージドテーブルs内で exactly-once（厳密に1回）の処理を実現します。重要なのは、それらの保証が自動的に適用されなくなる箇所を理解し、パイプラインの境界に適切な保護策を追加できるようにすることです。

仕組み​

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;

AUTO CDC以外で独自のアップサート論理を書く場合は(稀ですが複雑なMerge条件で必要となることがあります)、安定したビジネスキーでキーを付け、例えばブラインドINSERTではなくorder_idキーのキーMERGE ... WHEN MATCHEDを2回適用しても安全にしてください。詳細については、 『AUTO CDC APIs:パイプラインによるチェンジデータキャプチャの簡素化』をご覧ください。

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

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

  • マテリアライズドビューにおける非決定論的変換は避けてください。 マテリアライズドビューは完全に、または段階的に再計算できるため、出力が入力 値 ではなく実行 タイミング に依存する関数は避けてください。例えば、 current_timestamp() を使って、書き込まれた後に固定されるべきビジネス価値をコンピュートしないでください。ソースイベントのTimestampを取るかパラメーターとして渡すと、再コンピュート時に同一の出力が得られます。
  • フル更新が安全に行われるように設計してください。 フル更新を行うとテーブルが削除され、最初から再計算されます。これは、すべてのアップストリームソースが完全な履歴を生成できる場合にのみ安全です。アップストリームソースが変更のローリングウィンドウのみを公開している場合、ダウンストリームの 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」を参照してください。

少なくとも1回は発生するソースの重複を排除​

ソースがレコードを複数回配信する可能性がある場合は、ダウンストリームで重複排除を行います。ウォーターマークを 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"])
    )

サンプルデータ上で異なって見える列からではなく、ソースの一意性コントラクトからそれらの列を選択してください。正当に繰り返される可能性のある列をIDとして扱うと、実際のイベントが破棄されます。ユーザーが同じ広告を2回クリックするのが一般的な例です。ユーザーと広告に基づいて重複排除を行うと、2回目のクリックは通知なしに破棄されます。

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

制限事項​

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

  • foreach_batch_sink およびカスタム外部書き込み。 Spark はバッチが少なくとも 1 回は 試行 されることを保証しますが、部分的な書き込みの後に再試行されたバッチでは、一部の行が外部システムで 2 回表示される可能性があります。ナチュラルキーでのアップサートや、受信側で重複排除可能なバッチIDの書き込みなどを行い、外部書き込みをべき等にします。
  • シンクとしての Kafka。 Kafka トピックは Delta のようにトランザクションによる Exactly-Once 書き込みをサポートしていないため、Kafka へのマイクロバッチ書き込みが再試行されると、重複したメッセージが生成される可能性があります。ダウンストリームのコンシューマーが重複に敏感な場合は、イベント ID などを使用してコンシューマー側で重複排除を行ってください。
  • カスタムPythonのデータソースをソースとして使用しています。 読み取りが正確に一度だけかどうかは、ソース実装がオフセットから正しく報告・再開しているかどうかによります。オフセットを追跡しない場合は、少なくとも1回として扱い、イベントIDに基づいてdropDuplicatesを使用してダウンストリームで重複排除を行うか、AUTO CDCのキーベースのアップサートセマンティクスに依存します。

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

その他のリソース​

このページの見出し