PostgreSQL インジェストパイプラインを維持する

Important

Lakeflow Connect 用 PostgreSQL コネクタはパブリックプレビュー段階です。パブリックプレビューに登録する場合は、Databricks アカウントチームにお問い合わせください。

このページでは、PostgreSQL インジェストパイプラインを維持するための継続的な操作について説明します。

パイプラインの一般的なメンテナンス

このセクションのパイプラインメンテナンスタスクは、Lakeflow Connect のすべてのマネージドコネクタに適用されます。

一般的なパイプラインメンテナンスタスクについては、一般的なパイプラインメンテナンスタスクを参照してください。

未使用のステージングファイルを削除する

2025 年 1 月 6 日以降に作成したインジェストパイプラインの場合、ボリュームステージングデータは 25 日後に自動的に削除されるようにスケジュールされ、30 日後に物理的に削除されます。 25 日以上正常に完了していないインジェストパイプラインでは、宛先テーブル内にデータギャップが発生する可能性があります。ギャップを回避するには、ターゲットテーブルの完全な更新をトリガーします。

2025 年 1 月 6 日より前に作成されたインジェストパイプラインの場合、Databricks サポートに問い合わせて、ステージング CDC データの自動保持管理の手動有効化を依頼してください。

次のデータが自動的にクリーンアップされます。

CDC データファイル
スナップショットファイル
ステージングテーブルデータ

コネクタ固有のパイプラインメンテナンス

このセクションのパイプラインメンテナンスタスクは、PostgreSQL コネクタに固有のものです。

レプリケーションに新しいテーブルを追加する

既存のレプリケーションフローに新しいテーブルを追加するには:

レプリケーションユーザーに必要な特権を付与します。必要な特権の完全な一覧については、 PostgreSQL データベースのユーザー要件に関する説明を参照してください。
構造に基づいて、新しいテーブルのレプリカ ID を設定します。正しいレプリカ ID 設定の選択に関するガイダンスについては、テーブルのレプリカ ID の設定を参照してください。

パブリケーションにテーブルを追加します。

ALTER PUBLICATION databricks_publication ADD TABLE schema_name.new_table;

インジェストパイプラインの構成を更新して、新しいテーブルを含めます。これを行うには、Azure Databricks UI を使用するか、Databricks Asset Bundles バンドルまたは CLI コマンドで ingestion_definition を更新します。
インジェストゲートウェイを再起動して、新しいテーブルを検出します。ゲートウェイは定期的に新しいテーブルをチェックしますが、ゲートウェイを再起動すると検出プロセスが高速化されます。

レプリケーションスロットをクリーンアップする

インジェストパイプラインを削除しても、** レプリケーションスロットはソース PostgreSQL データベースから自動的に削除されません**。未使用のレプリケーションスロットにより、Write-Ahead ログ (WAL) ファイルが蓄積され、ソースデータベースのディスク領域がいっぱいになる可能性があります。

すべてのレプリケーションスロットを一覧表示するには:

SELECT slot_name, slot_type, active, restart_lsn, pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained_wal
FROM pg_replication_slots;

不要になったレプリケーションスロットを削除するには:

SELECT pg_drop_replication_slot('slot_name');

インライン DDL トラッキングを整理する

インライン DDL 追跡を無効にする場合は、各データベースに対して以下の手順を実行して、監査スクリプトによって作成されたオブジェクトをクリーンアップします。

イベントトリガーを削除します。

DROP EVENT TRIGGER IF EXISTS lakeflow_ddl_trigger CASCADE;

パブリケーションから監査テーブルを削除します。

ALTER PUBLICATION databricks_publication DROP TABLE public.lakeflow_ddl_audit;

監査テーブルを削除します。

DROP TABLE IF EXISTS public.lakeflow_ddl_audit CASCADE;

レプリケーションスロットの監視

レプリケーションスロットの状態を監視して、それらがアクティブで WAL データを使用していることを確認します。

SELECT slot_name,
       active,
       wal_status,
       active_pid,
       restart_lsn,
       confirmed_flush_lsn,
       pg_current_wal_lsn() AS current_lsn,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS replication_lag
FROM pg_replication_slots
WHERE slot_name LIKE 'databricks%';

レプリケーションのラグ値が大きいと、次のいずれかの問題が示される場合があります。

インジェストゲートウェイは、ソースデータベースの変更に対応していません。
インジェストゲートウェイが長期間停止されました。
ゲートウェイとソースデータベース間のネットワーク接続の問題。

レプリケーションスロットが非アクティブ (active = false) で、対応するパイプラインが不要になっていることを確認した場合は、レプリケーションスロットを削除してリソースを解放します。レプリケーションスロットのクリーンアップを参照してください。

WAL ディスクの使用状況を監視する

Write-Ahead ログ (WAL) ディスクの使用状況を監視して、ディスク領域の問題を防ぎます。

SELECT pg_size_pretty(sum(size)) AS wal_size
FROM pg_ls_waldir();

特定のレプリケーションスロットの WAL リテンション期間を確認するには:

SELECT slot_name,
       active,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained_wal,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) AS pending_wal
FROM pg_replication_slots
WHERE slot_name = 'your_slot_name';

注

ソースのセットアップ中に max_slot_wal_keep_size が適切に構成されている場合 ( レプリケーションスロットの WAL リテンション期間の制限で推奨)、非アクティブなレプリケーションスロットは無制限の WAL 増加を引き起こしません。制限に達するとスロットは無効になり、データベースのクラッシュを防ぎます。

WAL ディスク使用率が高い場合は、次の手順を実行します。

インジェストゲートウェイが継続的に実行されていることを確認します。
ゲートウェイログで、WAL データの使用を妨げている可能性のあるエラーを確認します。
WAL リテンション期間 (PostgreSQL 13 以上) を制限するように max_slot_wal_keep_size を設定することを検討してください。
```
ALTER SYSTEM SET max_slot_wal_keep_size = '10GB';
SELECT pg_reload_conf();
```
Warnung

max_slot_wal_keep_size設定すると、WAL リテンション期間の制限を超えた場合にレプリケーションスロットが無効になり、すべてのテーブルを完全に更新する必要があります。

インジェストゲートウェイを再起動する

ソースデータベースの負荷を減らすため、インジェストゲートウェイは定期的に新しいテーブルのみをチェックします。ゲートウェイが新しいテーブルを検出するまでに最大 6 時間かかる場合があります。このプロセスを高速化する場合は、ゲートウェイを再起動します。

さらに、次の状況でゲートウェイを再起動します。

ソースデータベースの構成を変更しました。
ゲートウェイでエラーまたはパフォーマンスの問題が発生しています。

パブリケーションを更新する

レプリケーションに含めるテーブルを変更する必要がある場合:

-- Add a table to the publication
ALTER PUBLICATION databricks_publication ADD TABLE schema_name.table_name;

-- Remove a table from the publication
ALTER PUBLICATION databricks_publication DROP TABLE schema_name.table_name;

-- List all tables in a publication
SELECT schemaname, tablename
FROM pg_publication_tables
WHERE pubname = 'databricks_publication';

パブリケーションを更新した後、インジェストゲートウェイを再起動して変更を適用します。

フィードバック

このページはお役に立ちましたか?

Last updated on 2025-12-19

次の方法で共有

PostgreSQL インジェスト パイプラインを維持する

パイプラインの一般的なメンテナンス

未使用のステージング ファイルを削除する

コネクタ固有のパイプライン メンテナンス