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

スタンドアロン ストリーミングテーブルの REPLACE WHERE フロー

REPLACE WHEREフローは、テーブル履歴全体を再処理することなく、スタンドアロンストリーミングテーブルのターゲットサブセットを再計算し、上書きします。これらは、遅れて到着するデータ、アップストリームの再処理、スキーマ進化、およびバックフィルを処理します。

REPLACE WHEREフローでは、ターゲットテーブルに述語を定義します。述語に一致するすべての行が削除され、その同じ述語範囲のソースクエリーを再評価することで置き換えられます。述語に一致しない行は変更されません。

要件​

「次の場合に置換」フローには、次の要件があります。

  • Databricksでは、Unity CatalogとServerlessコンピュートの使用をお勧めします。増分更新は、Serverless コンピュートでのみサポートされています。

REPLACE WHEREフローの使用時期​

以下のシナリオでは、REPLACE WHEREフローを使用します。

  • ストリーミングセマンティクスを使用しない増分バッチ処理 : ウォーターマークなどのストリーミング概念を管理せずに、新しい行をバッチで処理します。
  • 選択的再処理 : 述語に一致する行のみを再計算し、それ以外の行は変更しません。
  • 標準的なマテリアライズドビュー機能を超えるシナリオ :
    • ソースよりも保持期間が長いターゲットテーブル
    • ディメンションテーブルの変更時に再計算を防ぐ
    • 履歴全体を再計算しないスキーマ進化

REPLACE WHEREフローを作成する​

CREATE OR REFRESH STREAMING TABLE とインラインで FLOW REPLACE WHERE 句を使用します。

SQL
CREATE OR REFRESH STREAMING TABLE orders_enriched
SCHEDULE EVERY 1 DAY
FLOW REPLACE WHERE date >= date_add(current_date(), -7) BY NAME
SELECT
  o.order_id,
  o.date,
  o.region,
  p.product_name,
  o.qty,
  o.price
FROM orders_fct o
JOIN product_dim p
  ON o.product_id = p.product_id;

更新中に、述語と一致するターゲットテーブル内のすべての行が削除され、同じ述語範囲に対してソースクエリーが再計算され、新しい結果が挿入されます。この例では、過去7日間のすべての行がorders_enrichedから削除され、ソースクエリーを使用して再計算されます。

代わりに、長い形式の CREATE FLOW 構文を使用して、ストリーミングテーブルと REPLACE WHERE フローを別々のステートメントで定義します。

SQL
CREATE STREAMING TABLE orders_enriched;

CREATE FLOW orders_enriched AS
INSERT INTO orders_enriched BY NAME
REPLACE WHERE date >= date_add(current_date(), -7)
SELECT
  o.order_id,
  o.date,
  o.region,
  p.product_name,
  o.qty,
  o.price
FROM orders_fct o
JOIN product_dim p
  ON o.product_id = p.product_id;

両方のフォームは同等です。完全なCREATE FLOWリファレンスについては、CREATE FLOW (パイプライン)を参照してください。

ソースクエリーに述語を追加する必要はありません。パイプラインエンジンは、ソースから読み取るときにそれを自動的に適用します。

注記

BY NAME が必要です。これにより、列が位置ではなく名前で照合されます。

ヒストリカルデータのバックフィル​

スケジュールされた更新以外でヒストリカルデータや修正された行をターゲットテーブルに書き込むには、ヒストリカルデータが保存されている場所に基づいて、次の2つのメカニズムのいずれかを選択します。

  • 述語の上書き : REFRESH STREAMING TABLE ... WHERE を使用して、フローのソースクエリーを 1 回限りの述語範囲で再実行します。ヒストリカルデータがインクリメンタルデータと同じソースからのものである場合に使用します。
  • DML ステートメント : フローをバイパスして、ターゲット テーブルに直接挿入します。ヒストリカルデータがインクリメンタルデータとは別のソースに存在する場合に使用します。

述語のオーバーライド​

テーブル定義を変更せずに、単一の更新に対してフローの REPLACE WHERE 述語を上書きします。述語の上書きは1回限りのもので、その更新にのみ適用され、後続のスケジュールされた更新には影響しません。

REFRESH STREAMING TABLEで WHERE節を使用します。

SQL
REFRESH STREAMING TABLE orders_enriched WHERE date BETWEEN '2020-01-01' AND '2024-12-31';

フローは、オーバーライド述語に一致する行のみを削除し、ソースから再計算します。フローの静的述語は変更されていないため、次回のスケジュール更新では元の述語が使用されます。

オーバーライドを SYNC または ASYNC と組み合わせることができます。たとえば、バックグラウンドで更新を起動し、すぐにパイプラインの更新へのリンクを返すには、次のようにします。

SQL
REFRESH STREAMING TABLE orders_enriched WHERE date >= date_add(current_date(), -30) ASYNC;
注記

WHERE句は、FLOW REPLACE WHERE句を使用して作成されたストリーミングテーブルでのみサポートされます。他のストリーミングテーブルで指定すると、エラーが返されます。更新 ( マテリアライズドビューまたはストリーミングテーブル )を参照してください。

DMLステートメント​

ターゲットテーブルで直接 DML ステートメントをランして、フローとは別のソースからの行をロードします。

SQL
INSERT INTO orders_enriched
SELECT *
FROM orders_enriched_legacy
WHERE date < '2025-01-01';

フル更新動作​

REPLACE WHEREフローの完全更新では、フローで定義された述語のみを使用してソースクエリーが再実行されます。定義された述語範囲外で述語またはDMLステートメントによって以前に挿入された行も含め、その述語に一致しないターゲット内のすべての行を完全に削除します。

警告

フル更新を行うと、既存のデータがすべてクリアされ、定義された述語のみを使用してフローが再実行されます。7日間の述語で1年間実行されているパイプラインの場合、完全更新を行うと、テーブルには直近7日間のデータのみが含まれるようになります。それより古い行はすべて完全に削除されます。

SQL
REFRESH STREAMING TABLE orders_enriched FULL;

テーブルのフル更新を防ぐには、テーブルプロパティ pipelines.reset.allowed を falseに設定します：

SQL
CREATE OR REFRESH STREAMING TABLE orders_enriched
  TBLPROPERTIES (pipelines.reset.allowed = 'false')
  FLOW REPLACE WHERE date >= date_add(current_date(), -7) BY NAME
  ...

増分更新​

REPLACE WHEREフローは可能な場合に増分更新を使用し、置換ウィンドウ全体を再計算するのではなく、最後の更新以降に変更されたソースデータのみを再処理します。増分更新には、Serverless コンピュートが必要です。

増分更新が適用される場合​

以下のすべてがtrueである必要があります：

  • この取り込みパイプラインはサーバレスコンピュートで実行されます。
  • クエリーパターンはサポートされています。 サポートされている演算子セットについては、「 インクリメンタル 更新」を参照してください。
  • 述語は、ソーステーブルのベース列を参照します。 集計関数やウィンドウ関数の出力のような派生値に対する述語をソースにプッシュすることはできないため、増分更新は無効になります。
  • 現在の置換ウィンドウ内の行を変更した外部 DML はありません。 現在のウィンドウ外の行を変更する DML は影響を受けません。
  • 現在の置換ウィンドウには、以前の述語で除外された行は含まれません。 述語を拡張して以前に処理されなかった範囲をカバーする場合、その 1 回の更新は完全な再計算にフォールバックします。後続の更新は、再度増分更新の対象となります。
  • 述語は確定的です。 rand()のような非決定論的な関数を使用する述語は、増分更新を無効にします。current_date()などの時間関数が許可されています。

フローの最初の更新は、常に完全な計算になります。条件が満たされない場合、その更新は現在の置換ウィンドウの完全な再計算にフォールバックします。

増分更新のベスト プラクティス​

REPLACE WHEREフローが増分更新の対象となるように、次のガイドラインに従ってください。

移動下限を使用する​

下限が変動する述語は、無期限に増分更新の対象となります。

SQL
FLOW REPLACE WHERE date >= date_add(current_date(), -7)

date BETWEEN date_add(current_date(), -7) AND current_date()のような変動する上限値は、ウィンドウを移動して以前に除外された行を含めることができ、その結果、完全再計算への一度限りのfallbackが発生します。

GROUP BYに述語列を含める​

集計するときは、 GROUP BY に述語列を含めて、エンジンが集計の下に述語をプッシュできるようにします。

SQL
FLOW REPLACE WHERE date >= date_add(current_date(), -7) BY NAME
SELECT date, region, SUM(amount) AS total
FROM sales
GROUP BY date, region;

述語列が GROUP BY から欠落している場合、述語は集計の下にプッシュされず、ソース全体がスキャンされます。

述語列を結合キーに含める​

エンジンが結合されたすべてのソースをプルーニングできるように、述語列を結合条件に含めます。

SQL
FLOW REPLACE WHERE date >= date_add(current_date(), -7) BY NAME
SELECT f.date, f.user_id, d.region, f.revenue
FROM fact f
JOIN dim d ON f.date = d.date AND f.user_id = d.user_id;

結合されたテーブルが述語列を公開しない場合、そのテーブルは更新ごとに完全にスキャンされます。

フル再計算へのfallbackの診断​

更新が完全な再計算にフォールバックすると、フローの planning_information イベントに理由が報告されます。パイプライン イベント ログ スキーマの「IssueType オブジェクト」を参照してください。次の表は、イベントに報告された理由の一覧です。

理由:

意味

EXTERNAL_CHANGE_IN_REPLACE_WINDOW

外部 DML が現在の置換ウィンドウ内の行を変更しました。

REPLACE_WHERE_NOT_DETERMINISTIC

述語は非決定論的な式を使用します。

PRIOR_REPLACE_WHERE_NOT_DETERMINISTIC

前の更新では、非決定論的述語が使用されていました。

PRIOR_REPLACE_WHERE_PREDICATE_MISSING

前の更新で使用された述語を取得できませんでした。これは、その更新中に述語が正しくシリアル化されなかった場合に発生する可能性があります。

STALE_TARGET_DATA_IN_WINDOW

述語は更新の間に変更され、現在のウィンドウには以前の述語で除外された行が含まれています。

CANNOT_PUSH_PREDICATE

述語は少なくとも1つのソースにプッシュダウンできません。

REPLACE_WHERE_BACKFILL

この更新は、1回限りの述語オーバーライドを使用してヒストリカルデータを投入または以前の書き込みを修正するバックフィルです。

理由:

意味

EXTERNAL_CHANGE_IN_REPLACE_WINDOW

外部 DML が現在の置換ウィンドウ内の行を変更しました。

REPLACE_WHERE_NOT_DETERMINISTIC

述語は非決定論的な式を使用します。

PRIOR_REPLACE_WHERE_NOT_DETERMINISTIC

前の更新では、非決定論的述語が使用されていました。

PRIOR_REPLACE_WHERE_PREDICATE_MISSING

前の更新で使用された述語を取得できませんでした。これは、その更新中に述語が正しくシリアル化されなかった場合に発生する可能性があります。

STALE_TARGET_DATA_IN_WINDOW

述語は更新の間に変更され、現在のウィンドウには以前の述語で除外された行が含まれています。

CANNOT_PUSH_PREDICATE

述語は少なくとも1つのソースにプッシュダウンできません。

REPLACE_WHERE_BACKFILL

この更新は、1回限りの述語オーバーライドを使用してヒストリカルデータを投入または以前の書き込みを修正するバックフィルです。

例​

以下の例は、一般的なREPLACE WHEREフローパターンを示しています。

例 1: 保持期間が制限されたソースから履歴集計を保持する​

この例では、生データがソーステーブルから古くなった（3日間の保持）後でも、日次集計を無期限に保持します。

SQL
CREATE OR REFRESH STREAMING TABLE events_agg
FLOW REPLACE WHERE date >= date_add(current_date(), -3) BY NAME
SELECT
  date,
  key,
  SUM(val) AS agg
FROM events_raw
GROUP BY ALL;

例 2: ディメンション テーブルが変更されたときに再計算を防止する​

この例では、ディメンション属性が変更されたときに履歴ファクト行を変更せずに保持します。

SQL
CREATE OR REFRESH STREAMING TABLE fact_dim_join
FLOW REPLACE WHERE date >= date_add(current_date(), -1) BY NAME
SELECT
  f.date,
  f.user_id,
  d.region,
  f.revenue
FROM fact_table f
JOIN dim_users d
  ON f.user_id = d.user_id;

ユーザーの地域が変更された場合、直近の行のみが再計算されます。過去の行には、書き込まれた時点での地域の値が保持されます。

例3：完全な履歴を再計算せずに新しいメトリクスを追加する​

次の例は、テーブル定義を展開し、ターゲット範囲のみをバックフィルする方法を示しています：

  1. 初期テーブルを定義します。

    SQL
    CREATE OR REFRESH STREAMING TABLE clickstream_daily
    FLOW REPLACE WHERE event_date >= date_add(current_date(), -7) BY NAME
    SELECT
      event_date,
      page_id,
      COUNT(*) AS clicks
    FROM clickstream_raw
    GROUP BY ALL;

  2. クエリーを更新して uniq_users を追加します:

    SQL
    CREATE OR REFRESH STREAMING TABLE clickstream_daily
    FLOW REPLACE WHERE event_date >= date_add(current_date(), -7) BY NAME
    SELECT
      event_date,
      page_id,
      COUNT(*) AS clicks,
      COUNT(DISTINCT user_id) AS uniq_users
    FROM clickstream_raw
    GROUP BY ALL;

    7 日間のウィンドウより古い行には、uniq_users の NULL が含まれます。

例 4: 完全な履歴をバックフィルする前に小さいウィンドウで反復する​

この例では、過去の範囲全体を処理する前に、小さなデータウィンドウでクエリーロジックを検証する方法を示します。

短いウィンドウで起動してメトリクスを検証し、低いコンピュートコストでビジネスロジックを反復します：

SQL
CREATE OR REFRESH STREAMING TABLE revenue_attribution
FLOW REPLACE WHERE event_date >= date_add(current_date(), -7) BY NAME
SELECT
  event_date,
  campaign_id,
  SUM(revenue) AS total_revenue
FROM marketing_events
GROUP BY ALL;

短い期間を設定すると、各更新で直近7日間のみが再計算されるため、全履歴を対象に実行することをcommitする前に、必要な回数だけクエリーを修正できます。

クエリーが完成したら、DMLを使用して完全なヒストリカル範囲をバックフィルします。

SQL
INSERT INTO revenue_attribution
SELECT
  event_date,
  campaign_id,
  SUM(revenue) AS total_revenue
FROM marketing_events
WHERE event_date < date_add(current_date(), -7)
GROUP BY ALL;
このページの見出し