REPLACE USING フローによる部分的なスナップショットの置換
ベータ版
この機能はベータ版です。
REPLACE USING フローは、ターゲットテーブルをストリーミングソースと同期させます。指定されたキーカラムに一致するすべての行を置き換え、他のすべてのデータは変更しません。
SEQUENCE BY 列は更新を順序付けるため、更新が順不同で到着した場合でも結果は正しくなります。各キーに対して最も高いシーケンスが優先され、シーケンスの低い行がターゲット内に既に存在する高いシーケンスの行を上書きすることはありません。同じキーとシーケンスを共有する行は、置換されるのではなく追加されます。
REPLACE USINGの仕組み
2つのリージョンのクリックイベントとコンバージョンイベントを保持し、seqでシーケンス化されたeventsテーブルを検討します：
region_id | device_type | event_type | seq |
|---|---|---|---|
1 | iOS | をクリックします | 1 |
1 | Android | 変換 | 1 |
2 | iOS | をクリックします | 1 |
2 | デスクトップ | をクリックします | 1 |
REPLACE USING (region_id) SEQUENCE BY seq フローは、リージョン 1 および 3 についてこれらの更新を受信します。リージョン 2 には更新はありません：
region_id | device_type | event_type | seq |
|---|---|---|---|
1 | iOS | をクリックします | 2 |
1 | Android | 変換 | 2 |
1 | デスクトップ | をクリックします | 2 |
3 | iOS | をクリックします | 1 |
3 | デスクトップ | をクリックします | 2 |
ターゲットは次のようになります：
region_id | device_type | event_type | seq | 結果 |
|---|---|---|---|---|
1 | iOS | をクリックします | 2 | seq 2 が seq 1 よりも大きいため、置換されました |
1 | Android | 変換 | 2 | seq 2 が seq 1 よりも大きいため、置換されました |
1 | デスクトップ | をクリックします | 2 | seq 2 が seq 1 よりも大きいため、置換されました |
2 | iOS | をクリックします | 1 | この更新にはキーが含まれていないため、変更されていません |
2 | デスクトップ | をクリックします | 1 | この更新にはキーが含まれていないため、変更されていません |
3 | デスクトップ | をクリックします | 2 | 追加済み。キーに対しては最も高いシーケンスのみが適用されるため、リージョン 3 の seq 1 行は追加されません。 |
要件
「次を使用して置換」フローには、以下の要件があります。
- Databricks Runtime 18.2 以降で実行される REPLACE USING フロー（クラシックまたは Serverless コンピュート上）。Databricks では Unity Catalog の使用をお勧めします。
- ソースはストリーミングソースである必要があります。REPLACE USING は、ストリーミングではないソースを拒否します。
- 少なくとも1つのキー列と、1つの
SEQUENCE BY列を正確に指定する必要があります。
REPLACE USING フローの使用時期
LakeFlow Pipelinesには、既存の行を上書きする 3 つのフローが用意されています。ソースの形式と、置換対象の行を識別する方法に基づいて選択してください：
- ソースが列でキー指定された一連の部分的なスナップショットである場合は、 REPLACE USINGを使用してください 。REPLACE USINGは、受信データと一致するデータのみを上書きし、その他のデータはそのまま残します。プライマリーキーは必要ありません。
- AUTO CDCを使用するのは、 ソースが明示的な 挿入 、 更新 、 削除 操作を伴うチェンジデータキャプチャ (CDC) フィードである場合、または slowly changing dimension (SCD) タイプ2 の履歴が必要な場合です。AUTO CDCには、真の主キーも必要です。「AUTO CDC APIs: パイプラインによるチェンジデータキャプチャの簡素化」を参照してください。
- REPLACE WHERE を使用する場合: ソースがスナップショットであり、述語によって選択されたターゲットテーブルの範囲（例：過去 7 日間）をバッチ操作として再計算および上書きしたい場合。プライマリーキーは必要ありません。REPLACE WHERE フローを使用したバッチ処理を参照してください。
REPLACE USING フローを作成する
SQL または Python のいずれかで REPLACE USING フローを定義します。
- SQL
- Python
CREATE STREAMING TABLE とインラインで FLOW REPLACE USING 句を使用します：
CREATE STREAMING TABLE payments_current
FLOW REPLACE USING (payment_id) SEQUENCE BY payment_date BY NAME
SELECT payment_id, booking_id, status, payment_date
FROM STREAM(samples.wanderbricks.payments);
または、ロング形式のCREATE FLOW構文を使用します：
CREATE STREAMING TABLE payments_current;
CREATE FLOW payments_flow AS
INSERT INTO payments_current BY NAME
REPLACE USING (payment_id) SEQUENCE BY payment_date
SELECT payment_id, booking_id, status, payment_date
FROM STREAM(samples.wanderbricks.payments);
BY NAME はSQLで必要です。これは、位置ではなく名前で列を一致させます。
@dp.table を使用して、テーブルとフローを一緒に宣言します。
from pyspark import pipelines as dp
@dp.table(name="payments_current", replace_using=["payment_id"], sequence_by="payment_date")
def payments_current():
return spark.readStream.table("samples.wanderbricks.payments")
あるいは、@dp.replace_flow を使用して既存のストリーミングテーブルをターゲットにします。
from pyspark import pipelines as dp
dp.create_streaming_table("payments_current")
@dp.replace_flow(target="payments_current", replace_using=["payment_id"], sequence_by="payment_date")
def payments_flow():
return spark.readStream.table("samples.wanderbricks.payments")
replace_using はキー列のリストです。sequence_by は列名または Column 式であり、replace_using が設定されている場合は必須です。
シーケンスと順序どおりではないデータ
SEQUENCE BY 列により、更新が到着する順序に関係なく結果が一定になります。行のシーケンスがそのキーに対して既に保存されているシーケンスより大きい場合にのみ、その行がキーに適用されます。そのため、現在の値より古い遅延行や再送行は無視されます。更新時に存在しないキーは変更されません。
置換が予測どおりに動作するように、以下の慣行に従ってください。
実践 | 理由: |
|---|---|
Timestamp、バージョン番号、ログオフセットなど、キーバージョンごとに厳密に増加するシーケンスを使用してください。 | 同じキーとシーケンスを持つ2つの行が両方とも保持されるため、そのキーに対して重複する行が発生します。 |
Null 以外のシーケンスを使用してください。 | NULL シーケンスは、未定義の動作を引き起こす可能性があります。 |
エクスペクテーション
REPLACE USINGフローは期待値をサポートしています。warnとfailは他のフローと同様に動作します。warnは違反した行を保持して違反を記録し、failは更新を停止します。「パイプラインの期待値によるデータ品質の管理」を参照してください。
drop エクスペクテーションは、違反している行を、ソースが生成しなかったものとして扱います。削除された行は、ターゲットテーブル内の対応するキーを置換、削除、または変更しません：
- ドロップは重複排除の前に行われるため、フローはキーの最新の有効なバージョンを保持します。
- キーに対するすべての入力行がドロップされた場合、そのキーの既存の行は変更されません。
- 削除された行はシーケンスの下限を設定しないため、その後の有効な更新は、シーケンスが削除された行のシーケンスより低い場合でも適用されます。
制限事項
「REPLACE USING」フローには、以下の制限があります：
- REPLACE USING は、ターゲットテーブルごとに単一のフローをサポートしています。同じターゲット上で REPLACE USING を別のフロータイプと組み合わせることはサポートされていません。
- ターゲットテーブルは、パイプライン内に作成する必要があります。
- ソースはストリーミングソースである必要があります。
- 少なくとも1つのキー列と
SEQUENCE BY列を指定する必要があります。キー列を重複させることはできず、各キー列の型はソート可能である必要があります。整数、文字列、日付などのアトミック型はキーにできますが、MAPおよびVARIANTはキーにできません。
例
以下の例では、すべての Unity Catalog 対応ワークスペースで利用可能な、予約状態変更のサンプルテーブルである samples.wanderbricks.booking_updates から読み取ります。各予約は変更ごとに 1 回表示されるため、booking_id は新しい booking_update_id で繰り返されます。Wanderbricks データセットを参照してください。
例 1: キーごとに最新のレコードを保持する
各予約の現在の状態のみを保持します。フローは booking_id でキー設定され、booking_update_id でシーケンス化されるため、予約の最新の更新がそれ以前の更新を置き換えます。ソースが明示的な挿入、更新、削除操作を含むチェンジフィードである場合は、代わりに AUTO CDC を使用してください。
- SQL
- Python
CREATE OR REFRESH STREAMING TABLE bookings_current
FLOW REPLACE USING (booking_id) SEQUENCE BY booking_update_id BY NAME
SELECT booking_id, status, total_amount, booking_update_id
FROM STREAM(samples.wanderbricks.booking_updates);
from pyspark import pipelines as dp
@dp.table(
name="bookings_current",
replace_using=["booking_id"],
sequence_by="booking_update_id"
)
def bookings_current():
return spark.readStream.table("samples.wanderbricks.booking_updates")
同じ予約に対する複数の更新がTimestampを共有する可能性があるため、この例ではupdated_atTimestampではなくbooking_update_idでシーケンス処理を行います。シーケンスで同順位の行は置換されずに追加されるため、それらの予約については複数の行が残ることになります。
例 2: 複数の列にまたがるキー
レコードが列の組み合わせによって識別される場合は、それらすべてを REPLACE USING にリストしてください。ここでは各予約が (property_id, booking_id) で識別されるため、フローはプロパティごとの各予約の現在の状態を保持します。キー列が Null になる可能性がある場合、REPLACE USING は行をスキップするのではなく、Null と Null を一致させます。
- SQL
- Python
CREATE OR REFRESH STREAMING TABLE bookings_by_property
FLOW REPLACE USING (property_id, booking_id) SEQUENCE BY booking_update_id BY NAME
SELECT property_id, booking_id, status, total_amount, booking_update_id
FROM STREAM(samples.wanderbricks.booking_updates);
from pyspark import pipelines as dp
@dp.table(
name="bookings_by_property",
replace_using=["property_id", "booking_id"],
sequence_by="booking_update_id"
)
def bookings_by_property():
return spark.readStream.table("samples.wanderbricks.booking_updates")
例 3: エクスペクテーションを使用して無効なレコードを削除する
ターゲットに不正な行が含まれないように、エクスペクテーションを追加します。削除された行は、ソースが生成しなかったものとして扱われます。つまり、一致するキーを置換または削除することはなく、フローはそのキーの最新の有効な行にフォールバックします。このフローは、正の total_amount を持たない更新を削除します。
from pyspark import pipelines as dp
@dp.table(
name="bookings_validated",
replace_using=["booking_id"],
sequence_by="booking_update_id"
)
@dp.expect_or_drop("positive_amount", "total_amount > 0")
def bookings_validated():
return spark.readStream.table("samples.wanderbricks.booking_updates")