はじめに
こちらのクイックスタートをウォークスルーしました。
Lakebase変更データフィード(CDF: Change Data Feed)は、Lakebase Postgresテーブルに対する行レベルの変更(挿入、更新、削除)をUnity CatalogのDeltaテーブルとして永続化する機能です。OLTPデータベースであるLakebaseへの書き込みが、そのままレイクハウス側に変更履歴として蓄積されるので、監査証跡の確保や下流の分析パイプラインの構築が簡単になります。
Lakebase変更データフィード機能は執筆時点でパブリックプレビューです。
クイックスタートの流れは以下の4ステップです。
- 変更キャプチャを有効にする(REPLICA IDENTITY FULL)
- フィードを開始する
- 行をたどってレイクハウスに入る
- 行を変更し、流れを確認する
事前準備
Postgresデータベースを取得するを完了しており、playing_with_lakebaseサンプルテーブルを含むLakebaseプロジェクトがあることが前提となります。加えて、CREATE TABLE権限を持つUnity Catalogのカタログとスキーマが必要です。
私の環境では、Lakebaseプロジェクトmy projectのproductionブランチを使用します。ブランチのOverviewを開くと、Lakebase CDFというタブが追加されています。
この時点ではStatusはDisabledとなっています。
ステップ1: 変更キャプチャを有効にする
CDFが機能するには、Postgresのライトアヘッドログ(WAL)に完全な行データが記録されている必要があります。レプリカIDをFULLに設定することで、変更ごとに古い行の状態と新しい行の状態の両方が記録されるようになります。
Lakebase SQL Editorで以下を実行します。
ALTER TABLE playing_with_lakebase REPLICA IDENTITY FULL;
スキーマ内のすべてのテーブルに一括で設定する方法はこちらを参照してください。
ステップ2: フィードを開始する
Lakebase CDFはスキーマレベルで設定されます。ソーススキーマ内の現在および将来のすべてのテーブルが自動的に対象となるため、個々のテーブルを選択する必要はありません。
Lakebase CDFタブでStartをクリックすると、設定ダイアログが表示されます。ソースとしてデータベースdatabricks_postgresとスキーマpublicを、宛先としてUnity Catalogのカタログtakaakiyayoi_catalogとスキーマlakebase_cdfを選択しました。対象テーブルplaying_with_lakebaseがlb_playing_with_lakebase_historyという宛先テーブル名でWill includeと表示されています。
ダイアログにも警告が表示されていますが、CDFを開始するとコンピュートエンドポイントが再起動され、アクティブな接続が一時的に切断されます。本番環境で有効化する際にはタイミングにご注意ください。
なお、宛先にはDatabricks管理のDefault Storageを使用するカタログは指定できません。Free Editionのデフォルトのworkspaceカタログを指定しようとすると、以下のエラーが表示されます。
The selected catalog uses Databricks-managed Default Storage, which is not supported for Lakebase CDF. Please select a catalog backed by external storage.
外部ストレージを持つカタログを宛先として選択する必要があります。
Startをクリックすると最初のスナップショットがすぐに開始され、StatusがEnabledになります。SchemasタブにはソースとなるPostgresのスキーマと宛先のUnity Catalogスキーマのマッピングが表示されます。
ステップ3: 行をたどってレイクハウスに入る
まずはLakebase側で行を確認します。Lakebase SQL Editorでid=2の行を選択します。
SELECT * FROM playing_with_lakebase WHERE id = 2;
id=2の行はnameがc81e728d9d、valueが0.5769742となっています。
次に、Delta履歴テーブルで同じ行を探します。DatabricksワークスペースのSQLエディタ(SQLウェアハウス)に切り替えて実行します。
SELECT * FROM takaakiyayoi_catalog.lakebase_cdf.lb_playing_with_lakebase_history
WHERE id = 2;
Lakebase側と同じnameとvalueを持つ行が確認できますが、加えて以下のメタデータ列が付与されています。
-
_pg_change_type: 行がどのような種類のイベントを表しているか -
_timestamp: イベントの発生時刻 -
_pg_lsn、_pg_xid: Postgres側の順序情報
最初のスナップショットでは、既存のすべての行がinsertイベントとしてDeltaに書き込まれるため、この行の_pg_change_typeはinsertとなっています。宛先テーブルのスキーマ詳細は宛先テーブルスキーマとデータ型マッピングを参照してください。
ステップ4: 行を変更し、流れを確認する
Lakebase SQL Editorに戻り、id=2の行を更新します。
UPDATE playing_with_lakebase SET value = 55.5 WHERE id = 2;
変更がフィードに反映されるまで数秒待ってから、Databricks側で履歴テーブルを再度クエリーします。
SELECT id, value, _pg_change_type, _timestamp
FROM takaakiyayoi_catalog.lakebase_cdf.lb_playing_with_lakebase_history
WHERE id = 2
ORDER BY _pg_lsn DESC;
id=2の行が3回出現しています。
-
insert: 最初のスナップショットによる元の行(value = 0.5769742131233215) -
update_preimage: 更新前の古い値(value = 0.5769742131233215) -
update_postimage: 更新後の新しい値(value = 55.5)
_timestampを見ると、update_preimageとupdate_postimageは同一タイムスタンプで記録されており、UPDATE文の実行から数十秒程度でDelta側に反映されていました。行への変更がすべて新しい履歴行として追記されるので、常に完全な監査証跡が残ることになります。削除の場合も同様に_pg_change_type = 'delete'の行が1行追加されます。詳細は一般的な変更パターンを参照してください。
まとめ
REPLICA IDENTITY FULLの設定とGUIでの数クリックだけで、PostgresからDeltaテーブルへの変更データフィードを構成できました。スキーマレベルで設定するので、将来追加されるテーブルも自動的に対象となるのが運用上嬉しいポイントです。
履歴テーブルはそのままブロンズレイヤーとしてメダリオンアーキテクチャに組み込んだり、マテリアライズドビューや構造化ストリーミングで下流パイプラインを構築したりできます。本番運用にあたっては制限とトラブルシューティング、スキーマ変更管理もご確認ください。