Snowflake Openflow Gen2 にて Aiven の PosgtreSQL をソースにデータ連携する手順

Posted at 2026-10-28

概要

Snowflake Openflow Gen 2 の PostgreSQL コネクターを使用し、Aiven 上の PostgreSQL から Snowflake にデータを連携する手順を紹介します。

本記事では、Aiven で PostgreSQL と検証用テーブルを準備し、Snowflake 側にロール、データベース、デプロイメント、ランタイム、コネクターを作成します。Openflow の構成要素は主に SQL で作成し、コネクターの接続情報や同期対象テーブルは GUI で設定します。

また、初期データの取り込みに加え、PostgreSQL で実行した INSERT、UPDATE、DELETE、MERGE の変更が Snowflake に反映されることを確認します。一方、TRUNCATE TABLE による削除はコネクターでサポートされておらず、Snowflake に反映されないことも検証します。最後に、作成したリソースの削除手順を紹介します。

PostgreSQL の JDBC ドライバーを取得

PostgreSQL の JDBC ドライバーをダウンロード

Download | pgJDBC

出所: Download | pgJDBC

Aiven で PostgreSQL を準備

PostgreSQL を構築

Aiven のプロジェクト画面を開き、Create service を選択します。

サービスとして PostgreSQL を選択します。

Service tier で Free を選択し、Create service を選択します。

PostgreSQL サービスの作成が完了するまで待ちます。

Overview タブの Connection information に表示される Host と Password を控えます。これらは、後述する Snowflake Openflow の接続設定で使用します。

テーブルなどを準備

SQL editor タブを選択します。

以下の SQL を実行し、検証用の public.orders テーブルを作成します。あわせてレプリカ識別方式を設定し、初期データを登録します。

CREATE TABLE public.orders (
    order_id      INTEGER PRIMARY KEY,
    customer_name VARCHAR(100),
    amount        NUMERIC(12,2)
);

ALTER TABLE public.orders REPLICA IDENTITY DEFAULT;

INSERT INTO public.orders
    (order_id, customer_name, amount)
VALUES
    (1, 'Customer A', 100.00),
    (2, 'Customer B', 200.00);

SELECT * FROM public.orders;

変更データを論理レプリケーションで配信するため、Publication openflow_pub を作成し、public.orders を対象テーブルに追加します。

CREATE PUBLICATION openflow_pub
WITH (publish_via_partition_root = true);

ALTER PUBLICATION openflow_pub
ADD TABLE public.orders;

SELECT pubname, schemaname, tablename
FROM pg_publication_tables
WHERE pubname = 'openflow_pub';

Openflow の構築手順

ロールを準備

USE ROLE ACCOUNTADMIN;

以下の SQL の <現在のユーザー名> を、操作する Snowflake ユーザー名に置き換えてください。

CREATE ROLE IF NOT EXISTS OPENFLOW_ADMIN;
CREATE ROLE IF NOT EXISTS OPENFLOW_AIVEN_EXECUTE_RL;

GRANT ROLE OPENFLOW_ADMIN TO USER <現在のユーザー名>;
GRANT ROLE OPENFLOW_ADMIN TO ROLE ACCOUNTADMIN;
GRANT ROLE OPENFLOW_AIVEN_EXECUTE_RL TO ROLE OPENFLOW_ADMIN;

GRANT CREATE OPENFLOW DEPLOYMENT ON ACCOUNT TO ROLE OPENFLOW_ADMIN;
GRANT CREATE DATABASE ON ACCOUNT TO ROLE OPENFLOW_ADMIN;
GRANT CREATE INTEGRATION ON ACCOUNT TO ROLE OPENFLOW_ADMIN;
GRANT CREATE COMPUTE POOL ON ACCOUNT TO ROLE OPENFLOW_ADMIN;

ウェアハウスを準備

CREATE WAREHOUSE IF NOT EXISTS OPENFLOW_INGEST_WH
    WAREHOUSE_SIZE = 'XSMALL'
    AUTO_SUSPEND = 60
    AUTO_RESUME = TRUE;

GRANT USAGE, OPERATE ON WAREHOUSE OPENFLOW_INGEST_WH
    TO ROLE OPENFLOW_AIVEN_EXECUTE_RL;

データベースとスキーマを作成

USE ROLE OPENFLOW_ADMIN;

CREATE DATABASE IF NOT EXISTS OPENFLOW_DB;
CREATE SCHEMA IF NOT EXISTS OPENFLOW_DB.OPENFLOW_SCHEMA;

GRANT CREATE OPENFLOW RUNTIME
    ON SCHEMA OPENFLOW_DB.OPENFLOW_SCHEMA
    TO ROLE OPENFLOW_ADMIN;

GRANT CREATE OPENFLOW CONNECTOR
    ON SCHEMA OPENFLOW_DB.OPENFLOW_SCHEMA
    TO ROLE OPENFLOW_ADMIN;

GRANT USAGE ON DATABASE OPENFLOW_DB
    TO ROLE OPENFLOW_AIVEN_EXECUTE_RL;

GRANT USAGE ON SCHEMA OPENFLOW_DB.OPENFLOW_SCHEMA
    TO ROLE OPENFLOW_AIVEN_EXECUTE_RL;

CREATE DATABASE IF NOT EXISTS AIVEN_RAW;

GRANT USAGE, CREATE SCHEMA ON DATABASE AIVEN_RAW
    TO ROLE OPENFLOW_AIVEN_EXECUTE_RL;

GRANT USAGE ON DATABASE AIVEN_RAW
    TO ROLE OPENFLOW_AIVEN_EXECUTE_RL;

GRANT USAGE, CREATE TABLE ON SCHEMA AIVEN_RAW.PUBLIC
    TO ROLE OPENFLOW_AIVEN_EXECUTE_RL;

GRANT CREATE STREAM ON SCHEMA AIVEN_RAW.PUBLIC
    TO ROLE OPENFLOW_AIVEN_EXECUTE_RL;

Aiven の PostgreSQL のパスワードを登録

USE ROLE OPENFLOW_ADMIN;
USE DATABASE OPENFLOW_DB;
USE SCHEMA OPENFLOW_SCHEMA;

以下の SQL の <postgreSQLのパスワード> を、Aiven で確認した PostgreSQL のパスワードに置き換えてください。

CREATE SECRET AIVEN_PG_PASSWORD
    TYPE = GENERIC_STRING
    SECRET_STRING = '<postgreSQLのパスワード>';

GRANT READ ON SECRET AIVEN_PG_PASSWORD
    TO ROLE OPENFLOW_AIVEN_EXECUTE_RL;

ネットワークルールと外部アクセス統合を作成

以下の SQL の <postgreSQLのHost> を、Aiven で確認したホスト名に置き換えてください。ポート番号 22417 は本検証環境の値であるため、必要に応じて変更してください。

CREATE NETWORK RULE IF NOT EXISTS AIVEN_PG_NETWORK_RULE
    MODE = EGRESS
    TYPE = HOST_PORT
    VALUE_LIST = ('<postgreSQLのHost>:22417');

CREATE EXTERNAL ACCESS INTEGRATION IF NOT EXISTS AIVEN_PG_EAI
    ALLOWED_NETWORK_RULES = (
        OPENFLOW_DB.OPENFLOW_SCHEMA.AIVEN_PG_NETWORK_RULE
    )
    ENABLED = TRUE;

GRANT USAGE ON INTEGRATION AIVEN_PG_EAI
    TO ROLE OPENFLOW_AIVEN_EXECUTE_RL;

Openflow のデプロイメントを作成

USE ROLE OPENFLOW_ADMIN;
USE DATABASE OPENFLOW_DB;
USE SCHEMA OPENFLOW_SCHEMA;

CREATE OPENFLOW DEPLOYMENT AIVEN_DEMO_DEPLOYMENT
    DEPLOYMENT_TYPE = SNOWFLAKE
    DISPLAY_NAME = 'Aiven PostgreSQL Demo';

SELECT SYSTEM$WAIT_FOR_OPENFLOW_DEPLOYMENT_STATUS(
    600,
    'ACTIVE',
    'AIVEN_DEMO_DEPLOYMENT'
);

Openflow のデプロイメントを GUI で確認

Snowflake の画面で 取り込み → Openflow を選択します。

Openflow を起動 を選択します。

Deployments タブで、作成したデプロイメントが表示されていることを確認します。

Openflow のランタイムを作成

CREATE OPENFLOW RUNTIME AIVEN_DEMO_RUNTIME
    IN DEPLOYMENT AIVEN_DEMO_DEPLOYMENT
    NODE_TYPE = MEDIUM
    NODE_TYPE_TIER = 'M4'
    MIN_NODES = 1
    MAX_NODES = 1
    EXECUTE_AS_ROLE = OPENFLOW_AIVEN_EXECUTE_RL
    EXTERNAL_ACCESS_INTEGRATIONS = (AIVEN_PG_EAI)
    DISPLAY_NAME = 'Aiven PostgreSQL Runtime';

SELECT SYSTEM$WAIT_FOR_OPENFLOW_RUNTIME_STATUS(
    600,
    'ACTIVE',
    'OPENFLOW_DB.OPENFLOW_SCHEMA.AIVEN_DEMO_RUNTIME'
);

Openflow のランタイムを GUI で確認

Openflow の画面で Runtimes タブを開き、作成したランタイムが表示されていることを確認します。

Openflow のコネクターを作成

CREATE OPENFLOW CONNECTOR AIVEN_PG_CDC
    IN RUNTIME AIVEN_DEMO_RUNTIME
    FROM DEFINITION OPENFLOW_POSTGRES_CDC
    DISPLAY_NAME = 'Aiven PostgreSQL CDC';

SELECT SYSTEM$WAIT_FOR_STABLE_OPENFLOW_CONNECTORS(
    600,
    'OPENFLOW_DB.OPENFLOW_SCHEMA.AIVEN_PG_CDC'
);

Openflow のコネクターの設定を実施

Openflow の画面で Installed connectors タブを開き、作成したコネクターが表示されていることを確認します。

対象コネクターのケバブメニュー（縦三点メニュー）から Edit を選択します。

アクセス許可を求める画面で 許可 を選択します。

Source で以下の接続情報を設定し、Next を選択します。

Source Database Connection URL: jdbc:postgresql://:22417/defaultdb?sslmode=require
Source Database Driver: ダウンロードした JDBC ドライバーの JAR ファイルを選択
Source Database User: avnadmin
Source Database Password: AIVEN_PG_PASSWORD
Source Database Publication Name: openflow_pub

Source tables で同期対象の orders を選択し、Next を選択します。

Next を選択します。

以下の Snowflake 側の設定を指定し、Next を選択します。

Snowflake Destination Database: AIVEN_RAW
Snowflake Warehouse: OPENFLOW_INGEST_WH

Next を選択します。

Verifying configuration による設定の検証が完了することを確認します。

設定内容を適用するため、Apply を選択します。

Openflow のコネクターを起動

Openflow の画面で、対象コネクターのケバブメニューから Start を選択します。

再度ケバブメニューを開き、Monitor in Snowsight を選択して実行状況を確認します。

orders テーブルの同期が完了していることを確認します。

Snowflake 側で以下のクエリを実行し、AIVEN_RAW.PUBLIC.ORDERS に初期データが取り込まれていることを確認します。

SELECT
    *
FROM AIVEN_RAW.PUBLIC.ORDERS
ORDER BY ORDER_ID;