Azure Databricks へのインジェスト用に PostgreSQL を構成する

このページでは、Lakeflow Connect を使用して PostgreSQL から Azure Databricks に取り込むためのソース セットアップ タスクについて説明します。

セットアップと取り込み中に使用される資格情報

PostgreSQL インジェストでは、2 つの異なるステージで 2 つの異なる資格情報セットが使用されます。 使用する資格情報を知ることで、セットアップ中に認証とアクセス許可のエラーが発生するのを防ぎます。

変更データ キャプチャの論理レプリケーション

PostgreSQL コネクタは、論理レプリケーションを使用してソース テーブルの変更を追跡します。 論理レプリケーションを使用すると、ソース データベースに対するトリガーや大きなオーバーヘッドを必要とせずに、コネクタでデータの変更 (挿入、更新、削除) をキャプチャできます。

Lakeflow PostgreSQL 論理レプリケーションには、次のものが必要です。

1. Lakeflow Connect では、PostgreSQL バージョン 13 以降からのデータ レプリケーションがサポートされています。

2. 論理レプリケーション用にデータベースを構成します。

   PostgreSQL パラメーター wal_level を logicalに設定する必要があります。

3. レプリケートするすべてのテーブルを含むパブリケーションを作成します。

4. レプリケートされるカタログごとにレプリケーション スロットを作成します。

論理レプリケーションの詳細については、PostgreSQL Web サイトの 論理レプリケーション に関するドキュメントを参照してください。

ソースセットアップタスクの概要

Azure Databricks にデータを取り込む前に、PostgreSQL で次のタスクを完了します。

1. PostgreSQL 13 以降を確認する

2. ネットワーク アクセス (セキュリティ グループ、ファイアウォール規則、または VPN) を構成する

3. 論理レプリケーションを構成します。

   • 論理レプリケーションを有効にする (wal_level = logical)
   • 必要な特権を持つレプリケーション ユーザーを作成します。 PostgreSQL データベースのユーザー要件を確認する
   • テーブルのレプリカ ID を設定します。 テーブルのレプリカ ID の設定を参照してください
   • パブリケーションとレプリケーション スロットの作成

4. 省略可能: スキーマ変更の自動検出用にインライン DDL 追跡を構成します。 インライン DDL 追跡を選択する場合は、Databricks のサポートにお問い合わせください。

論理レプリケーションを構成する

PostgreSQL で論理レプリケーションを有効にするには、データベース設定を構成し、必要なオブジェクトを設定します。

WAL レベルを論理レベルに設定する

Write-Ahead ログ (WAL) は、論理レプリケーション用に構成する必要があります。 通常、この設定にはデータベースの再起動が必要です。

1. 現在の wal_level 設定を確認します。

   SHOW wal_level;
   

2. 値が logicalされていない場合は、サーバー構成で wal_level = logical 設定し、PostgreSQL サービスを再起動します。

レプリケーション ユーザーの作成

レプリケーション特権を使用して Databricks インジェスト用の専用 PostgreSQL ユーザーを作成します。

CREATE USER databricks_replication WITH PASSWORD 'your_secure_password';
GRANT CONNECT ON DATABASE your_database TO databricks_replication;
GRANT USAGE ON SCHEMA schema_name TO databricks_replication;
GRANT SELECT ON TABLE schema_name.table_name TO databricks_replication;
ALTER USER databricks_replication WITH REPLICATION;

特権の詳細な要件については、 PostgreSQL データベースのユーザー要件に関する記事を参照してください。

テーブルのレプリカ ID を設定する

レプリケートするテーブルごとに、レプリカ ID を構成します。 正しい設定は、テーブル構造によって異なります。

レプリカ ID の設定の詳細については、PostgreSQL ドキュメントの レプリカ ID を 参照してください。

パブリケーションを作成する

レプリケートするテーブルを含むパブリケーションを各データベースに作成します。 テーブルの所有者またはスーパーユーザーとして、次のコマンドを実行します。

-- Create a publication for specific tables
CREATE PUBLICATION databricks_publication FOR TABLE schema_name.table1, schema_name.table2;

-- Or create a publication for all tables in a database
CREATE PUBLICATION databricks_publication FOR ALL TABLES;

レプリケーション スロット のパラメーターを構成する

レプリケーション スロットを作成する前に、次のサーバー パラメーターを構成します。

レプリケーション スロットの WAL リテンション期間を制限する

パラメーター: max_slot_wal_keep_size

(既定値) にmax_slot_wal_keep_size-1。これにより、遅延または非アクティブなレプリケーション スロットによるリテンションにより無制限の WAL が肥大化する可能性があるためです。 ワークロードに応じて、このパラメーターを有限値に設定します。

max_slot_wal_keep_sizeパラメーターの詳細については、PostgreSQL の公式ドキュメントを参照してください。

レプリケーション スロットの容量を構成する

パラメーター: max_replication_slots

レプリケートされる各 PostgreSQL データベースには、1 つの論理レプリケーション スロットが必要です。 このパラメーターを、レプリケートするデータベースの数と、既存のレプリケーションのニーズに少なくとも設定します。

WAL 送信者を構成する

パラメーター: max_wal_senders

このパラメーターは、WAL データをサブスクライバーにストリーミングする同時 WAL 送信側プロセスの最大数を定義します。 ほとんどの場合、効率的で一貫性のあるデータ レプリケーションを確保するために、レプリケーション スロットごとに 1 つの WAL 送信側プロセスが必要です。

max_wal_sendersは、使用されているレプリケーション スロットの数と少なくとも同じになるように構成し、その他の既存の使用量を計算します。 運用の柔軟性を提供するために、少し高く設定することをお勧めします。

レプリケーション スロットを作成する

Databricks インジェスト ゲートウェイが変更の追跡に使用するレプリケーション スロットを各データベースに作成します。 レプリケーション スロットは、 REPLICATION 特権を持つユーザーによって作成される必要があります。 スーパーユーザーまたは管理者として接続している場合は、まずレプリケーション ユーザーに切り替えます。

SET ROLE databricks_replication;

-- Databricks supports only the pgoutput plugin for replication slots
SELECT pg_create_logical_replication_slot('databricks_slot', 'pgoutput');

-- Switch back to the admin or table owner role for subsequent steps
RESET ROLE;

省略可能: インライン DDL 追跡を構成する

インライン DDL 追跡は、コネクタがソース データベースからのスキーマ変更を自動的に検出して適用できるようにするオプションの機能です。 この機能は、既定では無効化されています。

どのスキーマ変更が自動的に処理され、どのスキーマが完全に更新される必要があるかについては、「 マネージド コネクタでスキーマの進化をどのように処理するか」 と 「スキーマの進化」を参照してください。

インライン DDL 追跡を設定する

下記のソース側設定を事前に完了すれば、ワークスペースで機能が有効になった時点でインラインDDLトラッキングがすぐに使えるようにします。 監査オブジェクトの作成は、機能が有効になるまでコネクタには影響がないため、事前にこれらのステップを実行しても安全です。

各PostgreSQLデータベースで以下のステップを完了します:

1. スクリプトの最新バージョンをダウンロードします。

   lakeflow_pg_ddl_change_tracking.sqlのダウンロード

2. スクリプトを実行します。

   \i lakeflow_pg_ddl_change_tracking.sql
   

   スクリプトは、 public スキーマに次のオブジェクトを作成します。 オブジェクト名には、スクリプトのバージョンを追跡するバージョン サフィックス (現在 _1_0) が含まれます。

   • 監査テーブル: public.lakeflow_ddl_audit_table_1_0 — キャプチャされた DDL イベントを格納します。
   • イベント トリガー関数: public.lakeflow_ddl_audit_function_1_0 ( ALTER TABLE イベントの場合) と public.lakeflow_drop_ddl_audit_function_1_0 ( DROP TABLE イベントの場合)。
   • イベント トリガー: lakeflow_ddl_audit_trigger_1_0 ( ddl_command_end で発生) と lakeflow_drop_ddl_audit_trigger_1_0 ( sql_dropで発生します)。

3. トリガーと監査テーブルが正常に作成されたことを確認します。

   -- Check for the DDL audit table
   SELECT * FROM pg_tables WHERE tablename LIKE 'lakeflow_ddl_audit_table%';
   
   -- Check for the event triggers
   SELECT * FROM pg_event_trigger WHERE evtname LIKE 'lakeflow%';
   

   監査テーブルの lakeflow_ddl_audit_table_1_0 と 2 つのイベント トリガー (lakeflow_ddl_audit_trigger_1_0 と lakeflow_drop_ddl_audit_trigger_1_0) が表示されます。

4. DDL 監査テーブルをパブリケーションに追加します。 このコマンドは、レプリケーション ユーザーではなくパブリケーション所有者として実行する必要があります。

   ALTER PUBLICATION databricks_publication ADD TABLE public.lakeflow_ddl_audit_table_1_0;
   

クラウド固有の構成に関する注意事項

AWS RDS と Aurora

• rds.logical_replication パラメーターがパラメーター グループ内で1に設定されていることを確認します。

• Databricks ワークスペースからの接続を許可するようにセキュリティ グループを構成します。

• レプリケーション ユーザーには、 rds_replication ロールが必要です。

  GRANT rds_replication TO databricks_replication;
  

Azure Database for PostgreSQL

• Azure portal または CLI を使用して、サーバー パラメーターで論理レプリケーションを有効にします。
• Databricks ワークスペースからの接続を許可するようにファイアウォール規則を構成します。
• フレキシブル サーバーでは、論理レプリケーションがサポートされています。 単一サーバーの場合は、サポートされているレベルを使用していることを確認します。

GCPのPostgreSQL用Cloud SQLサービス

• インスタンス設定で cloudsql.logical_decoding フラグを有効にします。
• Databricks ワークスペースからの接続を許可するように、承認されたネットワークを構成します。
• pglogical 拡張機能を使用している場合は、 cloudsql.enable_pglogical フラグが on に設定されていることを確認します。

構成を確認する

セットアップ タスクが完了したら、論理レプリケーションが正しく構成されていることを確認します。

1. wal_levelがlogicalに設定されていることを確認します。

   SHOW wal_level;
   

2. レプリケーション ユーザーが replication 特権を持っていることを確認します。

   SELECT rolname, rolreplication FROM pg_roles WHERE rolname = 'databricks_replication';
   

3. レプリケーション ユーザーがテーブルに対する SELECT 権限を持っていることを確認します。 schema_name.table_nameをレプリケートするスキーマとテーブルに置き換えます (たとえば、public.my_table)。

   SELECT has_table_privilege('databricks_replication', 'schema_name.table_name', 'SELECT');
   

4. パブリケーションが存在することを確認します。

   SELECT * FROM pg_publication WHERE pubname = 'databricks_publication';
   

5. レプリケーション スロットが存在することを確認します。

   SELECT slot_name, slot_type, active, restart_lsn
   FROM pg_replication_slots
   WHERE slot_name = 'databricks_slot';
   

6. テーブルのレプリカ ID を確認します。

   SELECT schemaname, tablename, relreplident
   FROM pg_tables t
   JOIN pg_class c ON t.tablename = c.relname
   WHERE schemaname = 'your_schema';
   

   relreplident列には、DEFAULT レプリカ ID のd (主キーを使用) または FULL レプリカ ID のf (主キーのないテーブルまたは TOASTable 列に必要) が表示されます。

次のステップ

ソースのセットアップが完了したら、インジェスト ゲートウェイとパイプラインを作成して PostgreSQL からデータを取り込むことができます。 PostgreSQL からのデータの取り込みを参照してください。