PostgreSQL 연결과 CDC

PostgreSQL 커넥터는 공식 JDBC 드라이버의 메이저 버전 42를 사용하며 PostgreSQL 8.2 이상과 호환됩니다. 원천 탐색, 배치 동기화, 증분 수집, Iceberg 동기화와 테이블 내보내기는 일반 제공(GA)입니다. CDC 동기화와 스트리밍 내보내기는 Beta 단계입니다.

소스와 인증

Data Connection의 +New Source에서 PostgreSQL 소스를 선택합니다. 사용자 이름과 비밀번호로 인증하며 개인 사용자보다 서비스 인증 정보가 권장됩니다. 사용자 이름과 비밀번호 인증에 클라이언트 또는 서버 인증서, 인증서를 검증하는 SSL 모드를 함께 사용할 수 있습니다.

사용자에게 대상 데이터베이스와 테이블의 읽기 또는 쓰기 권한이 필요합니다. CDC 구성에는 데이터베이스의 CREATE, REPLICATION 권한이 추가로 필요할 수 있습니다.

데이터베이스의 호스트 또는 IP에 접근할 수 있어야 하며 기본 포트는 5432입니다. Foundry 워커에는 대상의 외부 통신 정책을 추가합니다. 인터넷에서 접근 가능한 관리형 데이터베이스도 데이터베이스 측에서 Foundry의 유입 연결을 허용해야 합니다. Control Panel의 Network egress에서 Foundry 요청의 출발 IP를 확인합니다.

Amazon RDS에서는 AWS 콘솔에서 인스턴스의 실제 호스트 이름을 확인합니다. 정책의 호스트 형식은 <your-database-name>.<unique-identifier>.<region>.rds.amazonaws.com, 포트는 보통 5432입니다.

RDS에서는 인스턴스의 서버 인증서 체인을 신뢰하는 CA 인증서를 소스에 추가해야 할 수 있습니다. rds-ca-2019-root.pem 같은 기존 CA 파일을 다른 인증서 체인의 인스턴스에 그대로 적용하지 말고 실제 인스턴스에 사용 중인 CA와 맞춥니다.

연결 항목

항목	필수 여부	의미
`Host type`	필수	`Hostname`, `IPv4`, `IPv6` 중 연결 방식에 맞게 선택합니다. 호스트 이름이 권장되며 클라우드 호스팅 인스턴스에는 호스트 이름을 사용합니다.
`Port`	필수	연결 포트입니다. 보통 `5432`입니다.
`Database name`	필수	인스턴스 안의 데이터베이스 이름입니다.
`Authentication`	필수	사용자 이름과 비밀번호 및 필요한 인증서를 구성합니다.
`Network Connectivity`	필수	데이터베이스에 접근할 외부 통신 정책을 지정합니다.
`SSL Mode`	필수	기본값은 `verify-full`입니다.

논리 복제 기반 CDC

CDC는 논리 복제를 사용해 테이블의 행 변경을 거의 실시간으로 전달합니다. 복제 식별자는 보통 기본 키입니다. 기능이 제공되지 않으면 Palantir 지원에 접근을 요청할 수 있으며 Beta 단계에서 동작이 바뀔 수 있습니다.

소스는 기존 replication slot을 사용하거나 대상 데이터베이스에 slot과 publication을 생성합니다. 소스 하나에는 replication slot 하나와 publication 하나만 구성할 수 있지만 같은 연결로 여러 테이블을 스트리밍할 수 있습니다. 여러 slot이나 publication이 필요하면 같은 데이터베이스에 연결하는 소스를 각각 만듭니다.

먼저 일반 PostgreSQL 연결과 원천 탐색 실행 환경이 정상이어야 합니다. 초기화 중이면 몇 초 기다린 뒤 새로고침합니다. CDC syncs 탭에서 다음 항목을 지정합니다.

PostgreSQL CDC slot과 publication 설정

항목	필수 여부	의미
Replication slot name	필수	CDC에 사용할 slot 이름입니다. 없으면 생성합니다.
Publication name	필수	CDC에 사용할 publication 이름입니다.
Auto-create publication	필수	켜면 선택한 모든 테이블의 publication을 자동 생성합니다. 데이터베이스의 `CREATE`, `REPLICATION`과 테이블의 `SELECT` 권한이 필요합니다.

설정 후 Overview 또는 CDC syncs에서 + Create CDC sync를 선택하여 테이블별 동기화를 만듭니다.

외부 변환에서 읽기

다음 예제는 psycopg2로 plants 테이블을 질의하여 pandas 결과를 반환합니다. psycopg2, pandas 의존성이 필요합니다. 소스와 출력 데이터셋 RID, 호스트, 데이터베이스, 사용자 이름은 실제 구성에 맞게 바꾸고 소스에 PASSWORD라는 비밀 정보를 저장합니다. plants는 예시 테이블입니다.

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("<source_rid>")
)

@transform_pandas(output=Output("<output_dataset_rid>"))
def compute(postgres_source):

    conn = psycopg2.connect(
        host="host",            # Database host address
        port="5432",            # Database port
        database="postgres",    # Database name
        user="user",            # Database username
        password=postgres_source.get_secret("PASSWORD") # Database password
    )

    query = "SELECT * FROM plants;"

    return pd.read_sql_query(query, conn)

소스 UI의 인증과 네트워크 구성에 더해 코드에서 사용하는 데이터베이스 클라이언트의 연결 매개변수도 실제 환경에 맞아야 합니다. 필요한 서버 인증서와 TLS 검증 조건을 직접 연결 코드에도 적용합니다.

외부 변환에서 쓰기

다음 예제는 입력 데이터셋의 열 이름을 PostgreSQL 열 이름으로 매핑합니다. plant_name은 name, plant_height는 height_cm, plant_color는 color에 대응합니다. 입력에는 이 매핑으로 처리할 열만 있어야 합니다.

대상 plants 테이블이 없으면 생성하고 각 행을 삽입합니다. name이 기본 키이며 기존 키를 갱신하는 upsert는 수행하지 않습니다. 같은 키를 다시 삽입하면 데이터베이스 제약에 따라 실패할 수 있습니다. 실행하면 외부 데이터베이스가 변경되므로 대상과 입력을 확인하고 비운영 테이블에서 시험합니다.

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__)

# Maps Foundry column names to PostgreSQL column names
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)]})

쓰기 코드는 데이터베이스 트랜잭션 안에서 테이블 생성과 삽입을 수행하고 연결을 닫습니다. 오류가 발생하면 로그를 남기고 RuntimeError를 발생시킵니다. 성공 시 반환하는 rows_written은 입력 행 수이며 별도로 대상 테이블을 다시 조회한 결과는 아닙니다.

Keyboard shortcuts