Openflow Connector for PostgreSQL の設定¶

wal_levelの構成¶

Openflow Connector for PostgreSQL では wal_level を logical に設定する必要があります。

PostgreSQL サーバーがホストされている場所に応じて、wal_levelを以下のように構成できます：

パブリケーションの作成¶

Openflow Connector for PostgreSQL では、複製を開始する前に、 パブリケーション を PostgreSQL で作成し、構成する必要があります。すべてのテーブル、またはテーブルのサブセットに対して作成することも、特定の列のみを持つ特定のテーブルに対して作成することもできます。複製を予定しているすべてのテーブルと列がパブリケーションに含まれていることを確認してください。コネクタの実行中であれば、後でパブリケーションを変更することもできます。パブリケーションを作成および構成するには、以下の手順を実行します。

1. データベースに対する権限、を持つユーザーとしてログインし、次のクエリを実行します：

   • PostgreSQL 13以降の場合

     CREATE PUBLICATION <publication name> WITH (publish_via_partition_root = true);
     

     Copy

     パーティション分割されたテーブルを正しく複製するためには、 publish_via_partition_root を追加する必要があります。パーティション分割されたテーブルの取り込みについては、 パーティショニングされたテーブルを複製する をご参照ください。

   • PostgreSQL 13より前のバージョンの場合

     CREATE PUBLICATION <publication name>;
     

     Copy

2. 次を使用して、データベースエージェントが閲覧できるテーブルを定義します。

ALTER PUBLICATION <publication name> ADD TABLE <table name>;

Copy

パーティション分割されたテーブルの場合、ルートパーティションテーブルをパブリケーションに追加するだけで十分です。詳細については、 パーティショニングされたテーブルを複製する をご参照ください。

重要

PostgreSQL 15以降 では、指定されたテーブル列のサブセットに対するパブリケーションの構成をサポートしています。コネクタがこれを正しくサポートするには、 列フィルタリング設定 を使用して、パブリケーションで設定された列と同じ列を含める必要があります。

この設定がないと、コネクタは以下のように動作します：

• 宛先データベースでは、フィルターに含まれていない列に __DELETED というサフィックスが付きます。スナップショットフェーズ中にレプリケートされたすべてのデータは保持されます。

• パブリケーションに新しい列を追加すると、テーブルは永久に失敗し、複製を再起動する必要があります。

詳細については、 ALTER PUBLICATION をご参照ください。

フローパラメーター¶

まず、 PostgreSQL ソースパラメーターコンテキストのパラメーターを設定し、次に PostgreSQL 宛先パラメーターコンテキストのパラメーターを設定します。これが完了したら、コネクタを有効にすることができ、 PostgreSQL とSnowflakeの両方に接続して、実行が開始されることになります。しかし、明示的にテーブルが構成に追加されるまでは、いかなるデータも複製されません。

複製する特定のテーブルを構成するには、 PostgreSQL 取り込みパラメーターコンテキストを編集します。複製パラメーターコンテキストに変更を適用した直後に、コネクタが構成を取得し、各テーブルの複製ライフサイクルが開始します。

PostgreSQL レプリカサーバーからテーブルを複製する¶

コネクタは、論理的複製 を使用してプライマリサーバー、ホットスタンバイレプリカ、またはサブスクライバーサーバーからデータをインジェストできます。PostgreSQL レプリカに接続するコネクタを構成する前に、プライマリノードとレプリカノード間の複製が正しく機能することを確認します。コネクタでデータが欠落している問題を調査する場合は、まずコネクタが使用するレプリカサーバーに欠落している行が存在することを確認します。

スタンバイレプリカに接続する際のその他の考慮事項:

• ホットスタンバイレプリカへの接続のみがサポートされています。ウォームスタンバイレプリカは、プライマリインスタンスにプロモートされるまでクライアントからの接続を受け入れることができないことに注意してください。

• サーバーの PostgreSQL バージョンは>= 16である必要があります。

• コネクタが必要とする:ref:パブリケーション <label-postgres_connector_create_a_publication> は、スタンバイサーバーではなく、プライマリサーバーで作成する必要があります。スタンバイサーバーは読み取り専用で、パブリケーションの作成は許可されません。

ホットスタンバイインスタンスに接続して :ui:` 複製スロットを作成しようとすると、'<replication slot>' がタイムアウトします。スタンバイインスタンスに接続する場合は、プライマリ PostgreSQL インスタンスにトラフィックがあることを確認します。そうでなければ、複製スロットを作成するための呼び出しは`Openflow通知でエラーを返さないか、 Read PostgreSQL CDC Stream プロセッサーが起動しません。プライマリ PostgreSQL インスタンスにログインし、次のクエリを実行します。

SELECT pg_log_standby_snapshot();

エラーは、プライマリサーバーにデータ変更がない場合に発生します。そのため、レプリカサーバーで複製スロットを作成しているときに、コネクタが停止する可能性があります。これは、レプリカサーバーが複製スロットを作成できるように、プライマリサーバーから実行中のトランザクションに関する情報を要求するためです。プライマリサーバーはアイドル状態の間、情報を送信しません。pg_log_standby_snapshot() 関数は、プライマリサーバーが実行中のトランザクションに関する情報をレプリカサーバーに送信するように強制します。

テーブル複製の再開¶

（主キーが欠落している、またはスキーマの変更がサポートされていないなどの理由で） FAILED 状態のテーブルは、自動的に再開されません。テーブルが FAILED 状態になった場合、または複製を最初から再開する必要がある場合は、次の手順に従ってテーブルを削除し、複製に再度追加します。

1. フローパラメーターからテーブルを削除します。取り込みパラメーターのコンテキストで、Included Table Names からテーブルを削除するか、Included Table Regex を変更してテーブルが一致しないようにします。

2. テーブルが削除されたことを確認します。

   1. Openflowランタイムキャンバスで、プロセッサーグループを右クリックし、Controller Services を選択します。

   2. コントローラーサービスをリストしたテーブルで、Table State Store 行を見つけ、行の右側にある縦3つのドットをクリックして、View State を選択します。

   重要

   続行する前に、テーブルの状態がこのリストから完全に削除されるまで待つ必要があります。この構成変更が完了するまで続行しないでください。

3. 宛先をクリーンアップする:テーブルの状態が完全に削除されたと表示されたら、Snowflake で宛先テーブルを手動で DROP します。スナップショットフェーズ中に、コネクタは既存の宛先テーブルを上書きしないことに注意してください。テーブルがまだ存在する場合、複製は再度失敗します。オプションで、ジャーナルテーブルとストリームが不要になった場合は削除することもできます。

4. テーブルを再度追加する: Included Table Names または Included Table Regex パラメーターを更新して、テーブルを再度含めます。

5. 再開を確認する:前述の指示に従って Table State Store をチェックします。テーブルの状態は、ステータス NEW で表示され、次に SNAPSHOT_REPLICATION に移行し、最後に INCREMENTAL_REPLICATION になります。

テーブルの列のサブセットを複製します。¶

コネクタは、テーブルごとに複製されるデータを構成列のサブセットにフィルターできます。

列にフィルターを適用するには、複製パラメーターコンテキストのColumn Filterプロパティを変更し、フィルターを適用したいテーブルごとに1エントリずつ、構成の配列を追加します。

列は、名前またはパターン別に包含したり、除外したりすることができます。テーブルごとに単一の条件を適用することも、複数の条件を組み合わせて適用することもできます。除外は常に包含より優先されます。

以下の例は、利用可能なフィールドを示したものです。schema および table は必須で、 included、 excluded、 includedPattern、 excludedPattern のうちの1つ以上が必要です。

[
    {
        "schema": "<source table schema>",
        "table" : "<source table name>",
        "included": ["<column name>", "<column name>"],
        "excluded": ["<column name>", "<column name>"],
        "includedPattern": "<regular expression>",
        "excludedPattern": "<regular expression>",
    }
]

パーティショニングされたテーブルを複製する¶

このコネクタは、バージョン15以降の PostgreSQL サーバーでパーティション化されたテーブルの複製をサポートしています。パーティショニングされた PostgreSQL テーブルは、単一の宛先テーブルとしてSnowflakeに複製されます。

たとえば、パーティショニングされたテーブル orders およびサブパーティションの orders_2023 と orders_2024 があって、コネクタが orders.* パターンに一致するすべてのテーブルを取り込むように構成されている場合、Snowflakeには orders テーブルのみが複製され、そこにすべてのサブパーティションのデータが含まれることになります。

パーティショニングされたテーブルの複製をサポートするには、確実に PostgreSQL で作成された パブリケーション の publish_via_partition_root オプションが true に設定されているようにしてください。

パーティショニングされたテーブルの取り込みには、現在のところ以下の制限があります。

• 取り込み開始後に、パーティショニングされたテーブルにパーティションとしてテーブルがアタッチされると、コネクタはアタッチ前にパーティションテーブルに存在していたデータを取得しません。

• 取り込み開始後に、サブパーティションテーブルがパーティションテーブルから切り離されると、コネクタはルートパーティションテーブルでこのサブパーティションのデータを削除済みとしてマークしません。

• サブパーティションに対する切り捨て操作は、影響を受けるレコードを削除済みとしてマークしません。

テーブルでデータ変更を追跡する¶

コネクタはソーステーブルのデータの現在の状態だけでなく、すべての変更セットのすべての行の状態も複製します。このデータは、宛先テーブルと同じスキーマで作成されたジャーナルテーブルに格納されます。

ジャーナルテーブル名の形式は``<source_table_name>_JOURNAL_<timestamp>_<schema_generation>``です。ここで、``<timestamp>``はソーステーブルが複製に追加されたときのエポック秒の値であり、``<schema_generation>``はソーステーブルのスキーマ変更ごとに増加する整数です。その結果、スキーマが変更されるソーステーブルには、複数のジャーナルテーブルがあります。

テーブルが複製から削除され、再度追加されると、``<timestamp>``値が変更され、``<schema_generation>``は``1``から再開します。

コネクタはジャーナルテーブルをドロップすることはありませんが、複製されたすべてのソーステーブルに最新のジャーナルを使用し、ジャーナル上の追加専用ストリームのみを読み取ります。ストレージを回収するには、以下を実行します。

• すべてのジャーナルテーブルをいつでも切り捨てます。

• 複製から削除されたソーステーブルに関連するジャーナルテーブルをドロップします。

• アクティブに複製されたテーブルの最新の生成ジャーナルテーブルを除いて、すべてをドロップします。

例えば、コネクタがソーステーブル orders をアクティブに複製するように設定されており、以前にテーブル customers を複製から削除した場合、以下のようなジャーナルテーブルが存在する可能性があります。この場合、 orders_5678_2 を 除いて、それらのすべてをドロップできます。

customers_1234_1
customers_1234_2
orders_5678_1
orders_5678_2

マージタスクのスケジュールを構成する¶

コネクタはウェアハウスを使用して、変更データキャプチャ（CDC）データを宛先テーブルにマージします。この操作は、 MergeSnowflakeJournalTable プロセッサーによってトリガーされます。新しい変更がない場合、または MergeSnowflakeJournalTable キューで待機する新しいフローファイルがない場合、マージはトリガーされず、ウェアハウスは自動サスペンドします。

ウェアハウスのコストを制限し、スケジュールされた時間のみにマージを制限するには、Merge task Schedule CRON パラメーターで CRON 式を使用します。MergeSnowflakeJournalTable プロセッサーに送られてくるフローファイルをスロットルし、マージは専用の期間のみにトリガーされます。スケジュールに関する詳細は、 スケジュールストラテジー をご参照ください。

コネクタを停止または削除する¶

コネクタを停止または削除する場合は、コネクタが使用する 複製スロット を考慮する必要があります。

コネクタは、 snowflake_connector_ で始まる名前にランダムなサフィックスが続く独自の複製スロットを作成します。コネクタは複製ストリームを読み取るとスロットを進め、 PostgreSQL は WAL ログをトリミングしてディスク領域を解放します。

コネクタが一時停止している場合、スロットは進行せず、ソースデータベースへの変更によって WAL ログサイズが増加し続けます。特にトラフィックの多いデータベースでは、コネクタを長期間にわたり一時停止したままにしないでください。

コネクタをOpenflowキャンバスから削除したり、Openflowインスタンス全体を削除するなどの方法で削除しても、複製スロットはそのまま残るため、手動でドロップする必要があります。

同じ PostgreSQL データベースから複製する複数のコネクタインスタンスがある場合、各インスタンスは固有の名前の複製スロットを作成します。複製スロットを手動でドロップする場合は、それが正しいスロットであることを確認してください。CaptureChangePostgreSQL プロセッサーの状態を確認することで、指定されたコネクタインスタンスでどの複製スロットが使用されているかを確認できます。