PostgreSQL

Foundry を PostgreSQL ↗ に接続して、PostgreSQL データベースと Foundry の間でデータを読み取り、同期します。このコネクターは、メジャーバージョン42の公式 PostgreSQL ドライバー ↗を使用します。このドライバーは、PostgreSQL 8.2以降のすべてのバージョンに対応しています。

サポートされている機能

セットアップ

1. Data Connection アプリケーションを開き、画面右上の + 新規ソースを選択します。
2. 利用可能なコネクターの種類から PostgreSQL を選択します。
3. 追加の設定プロンプトに従い、以下のセクションの情報を使用してコネクターのセットアップを続けます。

Foundry でのコネクターのセットアップについて詳しくはこちら。

認証

Foundry の PostgreSQL コネクターでは、認証にユーザー名とパスワードを使用する必要があります。個々のユーザーの資格情報ではなく、サービスの資格情報を使用することを推奨します。

ユーザー名とパスワードによる認証は、クライアント証明書やサーバー証明書、およびこれらの証明書の検証を必要とする SSL モードと併用できます。

指定したユーザーが、対象のデータベースで必要な権限に加えて、対象のテーブルに対する読み取りまたは書き込み権限を持っていることを確認してください。変更データキャプチャでは、対象のデータベースに対する CREATE および REPLICATION 権限も必要になる場合があります。

ネットワーク

PostgreSQL コネクターには、接続先のデータベースインスタンスへのネットワークアクセスが必要です。PostgreSQL 接続では通常、ホスト名または IP アドレスを使用してポート5432に接続します。

PostgreSQL への接続を Foundry 内で実行する場合は、Data Connection アプリケーションでソースをセットアップするときに、適切なエグレスポリシーを追加する必要があります。

Amazon Relational Database Service (RDS) の PostgreSQL など、インターネット経由でアクセスできるクラウドホスト型の PostgreSQL インスタンスでは、データベースのホスト名に対するエグレスポリシーを追加する必要があります。ホスト名を使用しない場合は、IP アドレスに対するエグレスポリシーを追加する必要があります。必要なネットワーク設定の詳細については、マネージドのクラウドホスト型 PostgreSQL インスタンスのプロバイダーが提供する公式ドキュメントを確認してください。

Foundry から PostgreSQL インスタンスへの受信トラフィックが許可されていることを確認する必要があります。Foundry からのトラフィックの送信元となるエグレス IP アドレスは、Control Panel のネットワークのエグレスページで表示できます。これらの IP アドレスからデータベースインスタンスへのトラフィックを許可する方法については、ホスティングプロバイダーのドキュメントを確認してください。

Amazon RDS の PostgreSQL

Amazon RDS でホストされている PostgreSQL のマネージドインスタンスに接続する場合は、必要なエグレスポリシーを使用した直接接続を利用できます。

• Amazon RDS でホストされている PostgreSQL インスタンスのホスト名を確認するには、AWS コンソールで該当のインスタンスに移動します。RDS PostgreSQL のホスト名ベースのエグレスポリシーの例は、<your-database-name>.<unique-identifier>.<region>.rds.amazonaws.com (port 5432) です。

RDS PostgreSQL インスタンスに接続する際は、ソース設定パネルで RDS のルート認証局 (CA) をサーバー証明書として追加する必要がある場合があります。Amazon S3 サイト ↗ から rds-ca-2019-root.pem をダウンロードし、証明書の詳細を Foundry にコピーして、Amazon RDS への接続を信頼するようにします。SSL/TLS を使用した RDS データベースインスタンスへの接続の詳細については、公式の AWS ドキュメント ↗を確認してください。

接続情報

変更データキャプチャ [ベータ版]

PostgreSQL ソースは変更データキャプチャ（CDC）同期をサポートしています。

PostgreSQL は論理レプリケーションをサポートしているため、変更データキャプチャを使用すると、設定したテーブルの変更をほぼリアルタイムでストリーミングできます。PostgreSQL ドキュメント ↗ では、次のように説明されています。

論理レプリケーションは、レプリカアイデンティティ（通常は主キー）に基づいて、データオブジェクトとその変更を複製する方法です。

PostgreSQL からの Foundry の変更データキャプチャ同期は、既存のレプリケーションスロットを使用するか、対象データベースに新しいレプリケーションスロットとパブリケーションを作成することで動作します。Data Connection ソースごとに設定できるレプリケーションスロットとパブリケーションはそれぞれ1個のみですが、1つの接続で任意の数のテーブルを Foundry にストリーミングできます。複数のレプリケーションスロットまたはパブリケーションを使用する場合は、データベースに接続する Data Connection ソースを複数作成できます。

変更データキャプチャ同期を設定する前に、PostgreSQL ソース接続と探索ランタイムが正常に動作していることを確認してください。ランタイムがまだ初期化中の場合は、数秒待ってからページを再読み込みする必要がある場合があります。次に、CDC 同期タブに移動し、変更データキャプチャに必要な追加設定を指定します。

変更データキャプチャに必要な設定が完了したら、概要ページに移動するか、CDC 同期ページにとどまって、+ CDC 同期を作成を選択し、新しい変更データキャプチャ同期を作成できます。

PostgreSQL での CDC の使用について詳しくは、使用している PostgreSQL のバージョンの論理レプリケーションに関する公式ドキュメント ↗ を確認してください。

コードで PostgreSQL ソースを使用する

外部トランスフォームで PostgreSQL から読み取る

次の例では、詳細ガイド：初めての Data Connection を作成する ↗ ガイドで使用している Postgres ソースにクエリを実行します。

この例では、psycopg2 ↗ パッケージを使用して Postgres ソースにクエリを実行します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from transforms.api import transform_pandas, Output
from transforms.external.systems import external_systems, Source
import psycopg2
import pandas as pd


@external_systems(
    postgres_source=Source()
)

@transform_pandas(output=Output())
def compute(postgres_source):

    conn = psycopg2.connect(
        host="host",            # データベースのホストアドレス
        port="5432",            # データベースのポート
        database="postgres",    # データベース名
        user="user",            # データベースのユーザー名
        password=postgres_source.get_secret("PASSWORD") # データベースのパスワード
    )

    query = "SELECT * FROM plants;"

    return pd.read_sql_query(query, conn)

外部トランスフォームで PostgreSQL に書き込む

この例では、psycopg2 ↗ を使用して、入力データセットの行を PostgreSQL テーブルに書き込みます。対象テーブルがまだ存在しない場合は作成します。列マッピング辞書を使用して、Foundry の列名と PostgreSQL の列名の違いに対応します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
import logging
from transforms.api import lightweight, Output, transform_pandas, Input
from transforms.external.systems import external_systems, Source
import psycopg2
import pandas as pd

logger = logging.getLogger(__name__)

# Foundry の列名を PostgreSQL の列名に対応付けます
COLUMN_MAPPING = {
    "plant_name": "name",
    "plant_height": "height_cm",
    "plant_color": "color",
}

CREATE_TABLE_SQL = """
    CREATE TABLE IF NOT EXISTS plants (
        name TEXT PRIMARY KEY,
        height_cm DOUBLE PRECISION,
        color TEXT
    )
"""


@lightweight
@external_systems(
    postgres_source=Source("<source_rid>")
)
@transform_pandas(
    Output("<output_dataset_rid>"),
    plants_df=Input("<input_dataset_rid>")
)
def compute(postgres_source, plants_df):
    conn = psycopg2.connect(
        host="<your_host>",
        port="5432",
        database="<your_database>",
        user="<your_user>",
        password=postgres_source.get_secret("PASSWORD"),
    )

    pg_columns = [COLUMN_MAPPING[c] for c in plants_df.columns]
    placeholders = ", ".join(["%s"] * len(pg_columns))
    columns_str = ", ".join(pg_columns)
    insert_sql = f"INSERT INTO plants ({columns_str}) VALUES ({placeholders})"

    try:
        with conn:
            with conn.cursor() as cursor:
                cursor.execute(CREATE_TABLE_SQL)
                for _, row in plants_df.iterrows():
                    cursor.execute(insert_sql, tuple(row))
    except Exception as e:
        logger.error(f"Error writing to PostgreSQL plants table: {e}")
        raise RuntimeError(f"Failed to write data to PostgreSQL: {e}") from e
    finally:
        conn.close()

    return pd.DataFrame({"rows_written": [len(plants_df)]})