Postgres の変更をレイクハウスに格納する

Note

Lakebase Change Data Feed 機能は パブリック プレビュー段階です。

Postgres テーブルに Lakebase Change Data Feed (CDF) を設定し、行レベルの変更がコピー先の Delta テーブルに表示されるのを確認します。

手順: ① 変更キャプチャを有効にする → ② フィードを開始する → ③ 行がレイクハウスに取り込まれるのを追跡する → ④ 行を変更し、その変更が流れていくことを確認する

Note

これはクイックスタートです。 完全なドキュメントについては、「 Lakebase Change Data Feed」を参照してください。

始める前の準備

  • [ Postgres データベースの取得] が完了していることを確認します。 playing_with_lakebaseサンプル テーブルを含む Lakebase プロジェクトが必要です。
  • CREATE TABLE 権限がある Unity Catalog のカタログおよびスキーマ。

手順 1: 変更キャプチャを有効にする

Postgres では、CDF を機能させるために、先書きログに行の完全なデータが必要です。 レプリカ ID を完全に設定すると、すべての変更の古い行と新しい行の状態の両方を記録するように Postgres に指示されます。

Lakebase SQL エディターで、次のコマンドを実行します。

ALTER TABLE playing_with_lakebase REPLICA IDENTITY FULL;

詳細情報: スキーマ内のすべてのテーブルにレプリカ ID を設定し、新しいテーブルに自動適用する

手順 2: フィードを開始する

Lakebase CDF はスキーマ レベルで構成されます。 ソース スキーマのすべての現在および将来のテーブルは自動的に含まれるため、個々のテーブルは選択されません。

運用ブランチから、上部の階層リンクでブランチ名をクリックして ブランチの概要 を開き、[ Lakebase CDF ] タブを開き、[ スタート] をクリックします。 ソース スキーマとして public を選択し、移行先の Unity カタログ カタログとスキーマを選択します。 初期スナップショットがすぐに開始され、 lb_playing_with_lakebase_history は宛先に Delta テーブルとして表示されます。

ソースとコピー先の選択を含む [開始] ダイアログ。

詳細情報: 変更データ フィードを開始する

手順 3: 行をたどってレイクハウスに移動する

Lakebase から行を選択します。 行 id=2を見てみましょう。

SELECT * FROM playing_with_lakebase WHERE id = 2;

ここで、Delta 履歴テーブルで同じ行を見つけます。 Databricks SQL ウェアハウスまたはノートブックに切り替えて、次のコマンドを実行します。

SELECT * FROM <catalog>.<schema>.lb_playing_with_lakebase_history
WHERE id = 2;

<catalog>と<schema>を、手順 2 で選択した宛先に置き換えます。 Lakebase と同じid=2とnameを持つ行valueと、追加の列が表示されます。 初期スナップショットでは、既存のすべての行が insert イベントとして Delta に書き込まれます。これは、その行が表すものです。

これらの追加の列では、各行が表すイベントの種類 (_pg_change_type)、発生日時 (_timestamp)、Postgres の順序情報 (_pg_lsn、 _pg_xid) が示されます。

詳細情報: 変換先テーブル スキーマ | データ型マッピング

手順 4: 行を変更し、変更が反映されるのを確認する

Lakebase SQL エディターに戻り、行 id=2を更新します。

UPDATE playing_with_lakebase SET value = 55.5 WHERE id = 2;

変更がフィードに表示されるまで数秒待ってから、履歴テーブルのクエリを再実行します。

SELECT id, value, _pg_change_type, _timestamp
FROM <catalog>.<schema>.lb_playing_with_lakebase_history
WHERE id = 2
ORDER BY _pg_lsn DESC;

id=2 の 3 つの行を示す差分履歴テーブル: update_preimage、update_postimage、挿入

行 id=2 は、元の insert、古い値を持つ update_preimage 、新しい値を持つ update_postimage の 3 回表示されるようになりました。 行に対するすべての変更は新しい履歴行になるため、常に完全な監査証跡が作成されます。 削除は同じように機能し、1 つの行に _pg_change_type = 'delete'を追加します。

詳細情報: 一般的な変更パターン | ダウンストリーム パイプラインの構築

次のステップ

  • Last updated on