Lakebase 変更データ フィード

Lakebase Change Data Feed とは何ですか?

Lakebase では、ネイティブの Change Data Feed (CDF) が導入され、ダウンストリーム パイプライン、モデル、およびアプリケーションの運用データのロックが解除されます。 Lakebase Postgres テーブルに対するすべての挿入、更新、削除は、先書きログからキャプチャされ、Unity カタログマネージド Delta テーブルに新しい行として格納され、最大 15 秒ごとにバッチ処理およびフラッシュされます。 変更履歴は、任意のコンピューティング エンジンが読み取ることができるオープン形式で格納されます。

変換先テーブルは 、デルタ変更データ フィードと同じ図形に従います。各行には、 _pg_change_type、LSN、トランザクション ID、タイムスタンプが含まれます。 運用上の変更は、外部の CDC スタックを構築することなく、ETL、監査、ダウンストリームの利用者にとって主要なデータソースとなります。

ユースケース例

Lakebase CDF は、運用データを lakehouse に取り込むため、ダウンストリーム パイプラインとアプリケーションは、変更が発生した場合に対応できます。

このプレビューを有効にする

ワークスペース管理者は、ワークスペースのプレビュー ページから Lakebase Change Data Feedプレビューを有効にする必要があります。

Requirements

• レイクベースのオートスケーリング: Postgres 16、17、または18を動かす Lakebaseの自動拡大プロジェクト です。
• 出典データベース: ソーステーブルはLakebaseプロジェクトの任意の単一のデータベースに存在可能です。 CDFはフィードごとに1つのデータベースからの変更をキャプチャします。各プロジェクトが作成される databricks_postgres データベースに限定されません。
• Unity カタログ: CDF を構成する ID には、宛先カタログとスキーマに USE CATALOG、 USE SCHEMA、および CREATE TABLE が必要です。 オブジェクトに対する権限の付与を参照してください。
• 既定のストレージ: 既定のストレージで構成された宛先カタログはサポートされていません。
• Lakebase プロジェクト: Postgres ロールには、Lakebase プロジェクトに対する CAN MANAGE アクセス許可が必要です。 プロジェクト所有者には、既定で CAN MANAGE 権限があります。 「 プロジェクトのアクセス許可の管理」を参照してください。
• データ型:データ型のマッピングを参照してください。 Delta に直接対応するものがない型は、STRING として格納されます。

Lakebase CDF を設定する

開始するには、フィードで必要なテーブルにレプリカ ID を完全に設定し (手順 1)、Lakebase アプリで CDF を開始します (手順 2)。 データは、選択した Unity Catalog のカタログとスキーマに lb_<table_name>_history Delta テーブルとして表示されます。

手順 1: レプリカ ID を完全に設定する

Lakebase テーブルを CDF に参加させるには、 REPLICA IDENTITY FULL が設定されている必要があります。 既定では、Postgres では、行が更新または削除されたときに主キーのみがログに記録されます。 完全な ID を設定すると、事前書き込みログに前後の行の状態の両方を記録するように Postgres に指示されます。CDF は完全な変更履歴を作成する必要があります。

これらのコマンドは 、Lakebase SQL エディター または任意の Postgres クライアントで実行できます。

単一テーブル

ALTER TABLE <table_name> REPLICA IDENTITY FULL;

スキーマ内のすべての既存のテーブル

スキーマ内のすべての既存のテーブルにレプリカ ID を設定するには (この例ではpublic )、次のコマンドを実行します。

DO $$
DECLARE r record;
BEGIN
  FOR r IN
    SELECT table_schema, table_name
    FROM information_schema.tables
    WHERE table_schema = 'public'
      AND table_type = 'BASE TABLE'
  LOOP
    EXECUTE format(
      'ALTER TABLE %I.%I REPLICA IDENTITY FULL;',
      r.table_schema, r.table_name
    );
  END LOOP;
END $$;

将来のテーブルに自動適用する

新しく作成されたすべてのテーブルに REPLICA IDENTITY FULLを自動的に受信するには、 Postgres イベント トリガーをインストールします。 CREATE TABLEごとに実行され、新しいテーブルに ID が設定されます。

CREATE OR REPLACE FUNCTION public.set_full_replica_identity()
RETURNS event_trigger
LANGUAGE plpgsql
AS $$
DECLARE
  obj record;
BEGIN
  FOR obj IN
    SELECT * FROM pg_event_trigger_ddl_commands()
    WHERE command_tag = 'CREATE TABLE'
  LOOP
    EXECUTE format(
      'ALTER TABLE %s REPLICA IDENTITY FULL;',
      obj.object_identity
    );
  END LOOP;
END $$;

CREATE EVENT TRIGGER set_full_replica_identity_on_create
ON ddl_command_end
WHEN TAG IN ('CREATE TABLE')
EXECUTE FUNCTION public.set_full_replica_identity();

イベント トリガーを前のタブのループと組み合わせて、既存のテーブルと将来のテーブルの両方を 1 つのセットアップでカバーします。

SELECT n.nspname AS table_schema,
       c.relname AS table_name,
       CASE c.relreplident
         WHEN 'd' THEN 'default'
         WHEN 'n' THEN 'nothing'
         WHEN 'f' THEN 'full'
         WHEN 'i' THEN 'index'
       END AS replica_identity
FROM pg_class c
JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE c.relkind = 'r'
  AND n.nspname = 'public'
ORDER BY n.nspname, c.relname;

API:REPLICA IDENTITY FULL 標準的なPostgres DDLです。 PostgreSQLのALTER TABLE参照を参照してください。

手順 2: 変更データ フィードを開始する

Lakebase CDF はスキーマ レベルで構成されます。 開始すると、ソース スキーマ内のすべての現在および将来のテーブルがフィードに含まれます。

1. Azure Databricks ワークスペースで、アプリ スイッチャー (右上) から Lakebase Postgres を開きます。
2. Lakebase プロジェクトと、使用するブランチ ( 運用 、 メインなど) を選択します。
3. ブランチ の概要 を開くには、上部の階層リンクでブランチ名をクリックし、[ Lakebase CDF ] タブをクリックします。
4. [開始] をクリックします。
5. 構成ダイアログで、次の手順を実行します。
   • データベース: Source Postgresデータベースを選択してください。 プロジェクト内の任意のデータベースを選択でき、たとえプロジェクトにデータベースが1つだけであっても必ず1つを選択します。
   • スキーマ： ソース Postgres スキーマを選択します。
   • カタログへ: 移行先の Unity カタログ カタログを選択します。
   • スキーマ： コピー先の Unity カタログ スキーマを選択します。
6. [ 開始] をクリックしてフィードを開始します。

テーブルは lb_<table_name>_historyとして変換先に表示されます。 それらを見つけるには、サイドバーで カタログ を開き、コピー先のカタログとスキーマに移動して、[ テーブル ] タブを開きます。

[Lakebase CDF] タブには、次の 2 つのサブタブがあります。

• スキーマ： 各ソース スキーマ、Unity カタログ内のコピー先カタログとスキーマ、および状態を一覧表示します。
• テーブル： 各ソース テーブル、そのターゲット lb_<table_name>_history テーブル、状態 (Streaming または Snapshotting)、 コミット済み LSN (初期スナップショット中にフィードが Delta に書き込まれた距離( - として表示)、 最終更新 (テーブルが最後に変更された時刻) を一覧表示します。

Lakebase SQL エディターでこれを実行して、Postgres からのフィードの状態を検査することもできます。

SELECT * FROM wal2delta.tables;

結果には、テーブルごとの table_oid、 status (STREAMING または SNAPSHOTTING)、 committed_lsn、および last_write_time が含まれます。

API: フィードの設定やテーブルごとのステータスをプログラム的に取得するには、Lakebase Autoscaling APIガイドの 「Change Data Feed 」をご覧ください。

送信先テーブルのスキーマ

CDF は、ソース テーブルごとに 1 つの Delta テーブルを書き込みます。このテーブルは、ターゲット カタログとスキーマの lb_<table_name>_history という名前です。 ソース列に加えて、各行には次のシステム列が含まれます。

一般的な変更パターン

• 初期スナップショット: 既存の Lakebase テーブルで CDF を初めて実行すると、既存の各行が _pg_change_type = 'insert'で書き込まれます。
• 更新： 更新プログラムでは、 _pg_change_type = 'update_preimage' (古い行) と _pg_change_type = 'update_postimage' (新しい行) を含む行の 2 つの行が生成されます。
• 削除： 削除すると、 _pg_change_type = 'delete'を含む 1 つの行が生成されます。

これらは 差分変更データ フィードと同じ変更イベントであるため、同じダウンストリーム パターンが適用されます。

動作挙動

• 名前付けの競合: 2 つのソース テーブルが同じ宛先名にマップされる場合 (たとえば、 sales.users と marketing.users 両方の lb_users_historyへのマッピング)、CDF は最初のテーブルを lb_users_history に書き込み、2 番目のテーブルを lb_users_history_1に自動サフィックスを付けます。 Unity カタログでいずれかの変換先テーブルの名前を変更でき、フィードは引き続き機能します。
• スキーマ レベルのスコープ: Lakebase スキーマで CDF を開始すると、そのスキーマ内のすべての現在および将来のテーブルが含まれます。 空のテーブルはスキップされます。テーブルがコピー先に表示するには、少なくとも 1 つの行が必要です。
• 削除されたソース テーブル: Lakebase でテーブルを削除すると、Unity カタログの宛先 Delta テーブルが保持されます。

ダウンストリーム パイプラインを構築する

Lakebase CDF は、運用上の変更に対応するダウンストリーム パイプライン用に設計されています。 次のパターンは、フィードを使用する 3 つの方法を示しています。最も単純なものから最も柔軟なものまで順に並べられます。

シナリオの例。 eコマース アプリでは、Postgres orders テーブルに注文が記録され、各行に item_id と quantityが含まれます。 物流チームにはライブ 在庫レベルが必要です。 CDF では、 orders に対するすべての変更は、Unity カタログの lb_orders_history Delta テーブルに格納されます。 ダウンストリーム パイプラインは、注文が行われたり、編集されたり、キャンセルされたりするたびに、変更フィードを読み取り、 inventory_levels テーブルを更新します。

具体化されたビューを使用して現在のインベントリを計算する

最も簡単なパターンは、履歴テーブルに対する SQL 具体化されたビュー です。 MV は、新しい変更イベントが到着すると増分更新され、ダウンストリーム コンシューマーは他のテーブルと同様にクエリを実行します。

CREATE MATERIALIZED VIEW inventory_levels AS
SELECT
  item_id,
  SUM(
    CASE
      -- New orders (and the "new half" of updates) decrement inventory
      WHEN _pg_change_type IN ('insert', 'update_postimage') THEN -quantity
      -- Cancellations (and the "old half" of updates) restore inventory
      WHEN _pg_change_type IN ('delete', 'update_preimage') THEN quantity
      ELSE 0
    END
  ) AS current_inventory,
  MAX(_timestamp) AS last_transaction_ts,
  MAX(_pg_lsn) AS last_lsn
FROM lb_orders_history
GROUP BY item_id;

更新ごとに生成された 2 つの行は、ネット変更を除いて互いに取り消されるため、注文が編集されると、実行中の合計は正しいままです。

Spark 宣言型パイプラインを使用して変更をストリーム配信する

構造化された medallion アーキテクチャの場合は、 Lakeflow パイプラインを 使用して、ブロンズ、シルバー、ゴールドの各テーブルを宣言します。 Lakeflow パイプラインは、チェックポイントと依存関係管理が自動的に処理された接続されたパイプラインとして実行されます。

import dlt
from pyspark.sql import functions as F

@dlt.table
def inventory_adjustments():
    return (
        spark.readStream.table("<catalog>.<schema>.lb_orders_history")
        .withColumn(
            "delta",
            F.when(F.col("_pg_change_type").isin("insert", "update_postimage"), -F.col("quantity"))
             .when(F.col("_pg_change_type").isin("delete", "update_preimage"), F.col("quantity"))
             .otherwise(0),
        )
        .select("item_id", "delta", "_timestamp")
    )

@dlt.expect_or_drop("non_negative_stock", "on_hand >= 0")
@dlt.table
def inventory_levels():
    return (
        spark.read.table("LIVE.inventory_adjustments")
        .groupBy("item_id")
        .agg(F.sum("delta").alias("on_hand"))
    )

inventory_adjustmentsはlb_orders_historyを使用してreadStreamを増分的に読み取り、イベントごとの差分を生成します。 inventory_levels は、現在の在庫を計算するために item_id 別に集計されます。 想定では、在庫数がマイナスになる行は除外されます。これは上流側のバグを示しています。

完全なエンド ツー エンドのチュートリアルについては、「 チュートリアル: 変更データ キャプチャを使用して ETL パイプラインを構築する」を参照してください。

Spark 構造化ストリーミングを使用したカスタム処理

完全な制御 (カスタムマージ、副作用、複数のシンクなど) が必要な場合は、Spark 構造化ストリーミングを使用して履歴テーブルを直接読み取り、 foreachBatch を使用して宛先に書き込みます。

from pyspark.sql import functions as F
from delta.tables import DeltaTable

def update_inventory(batch_df, batch_id):
    deltas = (
        batch_df
        .withColumn(
            "delta",
            F.when(F.col("_pg_change_type").isin("insert", "update_postimage"), -F.col("quantity"))
             .when(F.col("_pg_change_type").isin("delete", "update_preimage"), F.col("quantity"))
             .otherwise(0),
        )
        .groupBy("item_id")
        .agg(F.sum("delta").alias("delta"))
    )

    target = DeltaTable.forName(spark, "<catalog>.<schema>.inventory_levels")
    (target.alias("t")
        .merge(deltas.alias("s"), "t.item_id = s.item_id")
        .whenMatchedUpdate(set={"on_hand": F.expr("t.on_hand + s.delta")})
        .whenNotMatchedInsert(values={"item_id": "s.item_id", "on_hand": "s.delta"})
        .execute())

(spark.readStream.table("<catalog>.<schema>.lb_orders_history")
    .writeStream
    .foreachBatch(update_inventory)
    .option("checkpointLocation", "/Volumes/<catalog>/<schema>/checkpoints/inventory_levels")
    .start())

各マイクロバッチは、 item_id によって変更イベントを集計し、ネットデルタを inventory_levelsにマージします。

段階的な拡張を前提とした設計。 各 lb_<table_name>_history テーブルは、追加専用の Delta テーブルです。 すべてのソースの変更は、操作を示す _pg_change_type とともに新しい行として記録されます。 Databricks SQL マテリアライズド ビュー、Lakeflow パイプライン フロー、Spark 構造化ストリーミング ジョブはすべて、Delta トランザクション ログから新たに追加された行を増分で処理するため、ダウンストリーム パイプラインは変更分に応じた処理のみを実行します。 変更セマンティクスは行データで既にエンコードされているため、履歴テーブルで 差分変更データ フィード を有効にする必要はありません。

データ型マッピング

CDF では、ほとんどの標準 PostgreSQL プリミティブ型がサポートされています。 Delta に直接対応するものがない型は、STRING として格納されます。

STRING として格納される型:

• Geography/Geometry (PostGIS): PostGIS 拡張機能の型 (たとえば、 geometry、 geography)。
• Vector (pgvector): pgvector 拡張機能からの vector 型。
• 複合型/構造体型:CREATE TYPE ... AS (field_name type, ...)で定義されたカスタム型。 これらは、名前付きフィールドを持つ行に似た型です。
• マップ：hstore などのマップに似たキー値型 (hstore拡張機能から)。 Postgres には組み込みのマップの種類がありません。 hstore は、キーと値のペアを列に格納する一般的な方法です。

スキーマ変更の管理

• Postgres のテーブルの名前を変更すると (たとえば、ALTER TABLE users RENAME TO customers)、フィードを続行できます。 変換先の Delta テーブル名は変更されず、 lb_users_historyのままです。
• スキーマの変更 (列の追加、列の削除、または列のデータ型の変更) によって、影響を受けるテーブルの再スナップショットがトリガーされます。 CDF は Postgres からテーブル全体を再読み取りし、コピー先の Delta テーブルに書き換えます。

Lakebase CDF を無効にする

CDF を無効にすると、プロジェクト内のすべての Lakebase スキーマのフィードが停止します。

1. Azure Databricks ワークスペースで、アプリ スイッチャー (右上) から Lakebase Postgres を開きます。
2. CDF を構成した Lakebase プロジェクトとブランチを選択します。
3. ブランチ の概要 を開くには、上部の階層リンクでブランチ名をクリックし、[ Lakebase CDF ] タブをクリックします。
4. [ 無効にする] をクリックします。 確認ダイアログで、変更がデルタ テーブルへのフローを停止するという警告を確認し、もう一度 [無効] をクリックして確認します。

CDF を無効にしても、コンピューティングは再起動されません。

API: フィード設定をプログラム的に無効化または削除するには、Lakebase Autoscaling APIガイドの「 データフィードの変更 」を参照してください。

制限事項とトラブルシューティング

テーブルごとの状態 (スナップショット、スキップ、またはストリーミング) は、 Lakebase の [CDF ] タブで確認するか、Lakebase で次を実行します。

SELECT * FROM wal2delta.tables;

テーブルがフィードに表示されない一般的な理由:

• REPLICA IDENTITY FULL 設定されていません: テーブルの ALTER TABLE <table_name> REPLICA IDENTITY FULL; を実行します。 「手順 1: レプリカ ID を完全に設定する」を参照してください。
• パーティション テーブル: Lakebase パーティション テーブルはサポートされていません。 パーティション テーブルを含むスキーマでは、これらのテーブルが失敗します。
• 空のテーブル: 行が 0 のテーブルは、少なくとも 1 つの行が存在するまでスキップされます。

次のステップ

• Spark 宣言パイプラインを使用して増分 ETL を構築します。 詳しい手順については、チュートリアル: 変更データ キャプチャを使用して ETL パイプラインを構築するを参照してください。
• Databricks SQL を使用してブロンズ レイヤーのクエリを実行します。 Databricks SQL を使用したデータウェアハウジングの開始方法を参照してください。
• 宛先の Delta テーブルに対するタイム トラベル クエリを含む監査履歴。