実際のパイプラインではリトライやリランは避けられないため、このページではLakeflowパイプラインが提供する処理保証と、書き直した部分を安全に再実行する方法を説明します。
Overview
パイプラインの再稼働が安全かどうかを決定する関連する性質は以下の通りです。
- 冪無能 とは、同じ入力に対して何度実行してもパイプラインが同じ結果を生むことを意味します。 失敗後に再実行したり、日付範囲を2回埋め戻したり、ジョブを手動で再トリガーしても、重複行が生んだり状態が破損したりすることは決してありません。
- 処理保証は 、各レコードが結果に何回影響するかを示します。 少なくとも一度の処理であれば すべてのレコードが処理されますが、失敗や再試行で一部のレコードが複数回処理され、重複のリスクがあります。 正確に一度 処理することで、すべてのレコードが一度だけ処理されたかのように結果に影響を与え、重複やギャップがなくても、再試行を繰り返しても効果が保証されます。
Lakeflow のパイプラインは、デフォルトでは管理対象のコンポーネントに対してべき等性を持ち、独自の管理対象テーブル内では「正確に 1 回」の処理を実現します。 重要なのは、その保証がどこで自動的になくなるのかを理解することで、パイプラインの端に適切な安全策を付けることができます。
どのように機能するのか
Lakeflowパイプラインは、管理するフローに対して正確に一度の処理と冪等性を提供し、書いた論理を冪等性に保つためのツールも提供します。
管理型テーブルの厳密に1回の処理
管理型テーブルでは、デフォルトで正確に一度だけの処理が可能です。 ストリーミングテーブルはStructured StreamingチェックポイントとDelta Lakeのトランザクション書き込みを組み合わせて使用します。各マイクロバッチはソースオフセットと出力を同時にコミットするため、失敗後の再試行バッチは完全に成功するか、完全にロールバックされて再試行され、部分的に二度適用されることはありません。 これは、Auto Loader によるファイル取り込み、Kafka、Kinesis、Azure Event Hubs からの読み取り、および AUTO CDC へのアップサートにも、コードを記述しなくても適用されます。
少なくとも一度は同じレコードを複数回送信した場合、パイプラインはそれらを一意のレコードとして処理し、すべてをテーブルに書き込みます。 重複を取り除くのはあなたの責任です。 少なくとも1回のデータソースの重複排除を参照してください。
読み取りの冪等性も、これらと同じチェックポイントによって成り立ちます。 オートローダーとストリーミング テーブルのチェックポイントにより、状態を追跡するために各ソースファイルまたはオフセットが一度だけ処理されることが保証されるため、障害発生後にパイプライン更新を再処理する場合は、データを再処理したりスキップしたりするのではなく、チェックポイントから再開されます。 これは、手作業で組んだバッチループではなく、spark.readStream 上のストリーミングテーブルを使うことで実現できます。
ストリーミング テーブルを参照してください。
手書きのMERGEではなくAUTO CDCを使え
AUTO CDC INTO はその keys と sequence_byに関して本質的に冪我性を持つ。 同じ変更レコードを2回適用したり、レコードを順番を外して適用しても、パイプラインはシーケンス列を使って、入力された行が実際に保存されている行より新しいかどうかを判断するため、同じ最終状態になります。
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 の外で独自のアップサート処理を実装する場合は（まれですが、複雑なマージ条件では必要になることがあります）、安定したビジネスキーを使い、2回適用しても安全なものにしてください。たとえば、無条件の INSERT ではなく、MERGE ... WHEN MATCHED をキーにした order_id を使用します。 詳細については、 『AUTO CDC API:パイプラインによる変更データキャプチャの簡素化』をご覧ください。
独自の変換をべき等に保つ
書き込み操作を再実行する際に論理を冪等性に保つために、以下の2つのガイドラインに従います。
- マテリアライズドビューでは、非決定的な変換を避けてください。 マテリアライズドビューは完全または段階的に再計算できるため、出力が実行 時間に依存して 入力が何である か に依存する関数は避けてください。 例えば、
current_timestamp()を使ってビジネス値を計算せず、書き込み後も固定されるべきです。元のイベントからタイムスタンプを取るか、パラメータとして渡して再計算時に同じ出力を出します。 - フルリフレッシュは安全に設計しましょう。 フルリフレッシュはテーブルを一から削除・再計算しますが、これはすべての上流ソースが完全な履歴を生成できる場合にのみ安全です。 上流ソースがローリングウィンドウの変更のみを公開する場合、下流の
AUTO CDCテーブルの完全な更新は静かに履歴を失う可能性があるため、ソースとトピックの保持をこの点を考慮して設計してください。
エッジで「正確に 1 回」を実現する
「正確に 1 回」が自動的ではなくなるのは、外部システムへの書き込みなど、パイプラインが直接制御する範囲の境界においてです。 外部システムに広げる際は、書き込み自体を冪等にしてください。例えば、受信側でキーでアップサートするなどです。なぜなら、再試行されたマイクロバッチは同じバッチを2回書く可能性があるからです。 以下のシンクはバッチの各パーティションを実行者から書き込み、冪等キーを使って再試行バッチが二重書きをしないようにします。
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 パイプラインの Sinksをご覧ください。
少なくとも1回のソースの重複を排除する
あるソースがレコードを複数回配信できる場合は、下流で重複を減らしましょう。 ウォーターマークと dropDuplicatesWithinWatermarkを組み合わせると、ウォーターマーク認識があり、重複検出に無制限状態を必要としません。 イベントを一意に識別する列の重複を減らします。 どの単一列もそれだけでは一意にならない場合、識別子は複数の列にわたることがあります。 以下の例では、クリックシーケンス番号はセッション内でのみ一意であるため、2つの列が合わせてイベントを識別します。
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"])
)
その列は、サンプルデータ上で一見異なって見えるものではなく、ソースで定義された一意性制約に基づいて選んでください。 正当に重複し得る列を識別子として扱うと、実際のイベントを取りこぼしてしまいます。 ユーザーが同じ広告を2回クリックする例がよくある例です。重複除去を行った場合、広告は静かに2回目のクリックを落とします。
AUTO CDCのキーベースのupsertセマンティクスも自然に重複を崩壊させるため、安定したビジネスキーを基にした AUTO CDC フローを通じて少なくとも一度データをルーティングする方法も、正確に一度の状態に収束する方法の一つです。
制限事項
厳密に1回限りの処理は、管理対象の Delta から Delta へのフローに適用されます。 以下のエッジは少なくとも1回実行されるものとして扱い、それらに明示的な重複排除または冪等書き込みロジックを追加します。
-
foreach_batch_sinkとカスタム外部書き込み。 Sparkはバッチが 少なくとも一度は試み られることを保証しますが、部分的な書き込み後に再試行すると、外部システム上で一部の行が2回表示されることがあります。 外部書き込みを冪等にし、例えば自然鍵にアップサートしたり、受信側が重複を解消できるバッチIDを書き込むなどです。 - シンクとしてのKafka。 Kafka のトピックは、Delta のようなトランザクションによる exactly-once の書き込みをサポートしていないため、再試行されたマイクロバッチを Kafka に書き込むと、重複メッセージが生成される可能性があります。 下流の消費者が重複に敏感であれば、消費者側で重複除去を行い、例えばイベントIDで重複を除去します。
- カスタムPythonデータソースをソースとして使用しています。 読み取りが「正確に 1 回」となるかどうかは、ソースの実装がオフセットを正しく報告し、そのオフセットから処理を再開できるかどうかに依存します。 オフセットを追跡しない場合は、それを「少なくとも 1 回」として扱い、イベント ID に基づいて
dropDuplicatesで下流で重複排除を行うか、AUTO CDCのキー ベースのアップサートセマンティクスに依存して処理してください。
大まかな目安として、パイプライン全体が Delta-to-Delta (ストリーミングテーブルやマテリアライズド ビューが、マネージドフローを通じて差分テーブルの読み書きを行う) である場合、すでに「正確に 1 回」が実現されています。
foreach_batch_sink、差分以外のシンク、または未検証のカスタムソースを追加した時点で、その特定のエッジを「少なくとも 1 回」として扱い、そこにべき等な書き込み、または重複排除のロジックを追加してください。
