Sink data from RisingWave to PostgreSQL

This guide will show you how to sink data from RisingWave to PostgreSQL using the JDBC connector. The sink parameters are similar to those for other JDBC-available databases, such as MySQL. However, we will cover the configurations specific to PostgreSQL and how to verify that data is successfully sunk.

You can test out this process on your own device by using the postgres-sink demo in the integration_test directory of the RisingWave repository.

​

Set up a PostgreSQL database

psql --host = pg-to-rw.xxxxxx.us-east-1.rds.amazonaws.com --port=5432 --username=awsuser --password

psql --host = pg-to-rw.xxxxxx.us-east-1.rds.amazonaws.com --port=5432 --username=awsuser --password

psql postgresql://myuser:123456@127.0.0.1:5432/mydb

GRANT SELECT, INSERT, UPDATE, DELETE ON [table_name] TO [username];

​

Create a table in PostgreSQL

Use the following query to set up a table in PostgreSQL. We will sink to this table from RisingWave.

CREATE TABLE target_count (
  target_id VARCHAR(128) PRIMARY KEY,
  target_count BIGINT
);

​

Set up RisingWave

​

Install and launch RisingWave

To install and start RisingWave locally, see the Get started guide. We recommend running RisingWave locally for testing purposes.

​

Notes about running RisingWave from binaries

If you are running RisingWave locally from binaries and intend to use the native CDC source connectors or the JDBC sink connector, make sure you have JDK 11 or later versions installed in your environment.

​

Create a sink

​

Syntax

CREATE SINK [ IF NOT EXISTS ] sink_name
[FROM sink_from | AS select_query]
WITH (
   connector = 'jdbc' | 'postgres',
   field_name = 'field', ...
);

​

Parameters (JDBC)

​

Parameters (Postgres Native)

​

Sink data from RisingWave to PostgreSQL

​

Create source and materialized view

You can sink data from a table or a materialized view in RisingWave to PostgreSQL.

For demonstration purposes, we’ll create a source and a materialized view, and then sink data from the materialized view. If you already have a table or materialized view to sink data from, you don’t need to perform this step.

Run the following query to create a source to read data from a Kafka broker.

CREATE SOURCE user_behaviors (
    user_id VARCHAR,
    target_id VARCHAR,
    target_type VARCHAR,
    event_timestamp TIMESTAMPTZ,
    behavior_type VARCHAR,
    parent_target_type VARCHAR,
    parent_target_id VARCHAR
) WITH (
    connector = 'kafka',
    topic = 'user_behaviors',
    properties.bootstrap.server = 'message_queue:29092',
    scan.startup.mode = 'earliest'
) FORMAT PLAIN ENCODE JSON;

Next, we will create a materialized view that queries the number of targets for each target_id. Note that the materialized view and the target table share the same schema.

CREATE MATERIALIZED VIEW target_count AS
SELECT
    target_id,
    COUNT(*) AS target_count
FROM
    user_behaviors
GROUP BY
    target_id;

​

Sink from RisingWave

Use the following query to sink data from the materialized view to the target table in PostgreSQL. Ensure that the jdbc_url is accurate and reflects the PostgreSQL database that you are connecting to. See CREATE SINK for more details.

CREATE SINK target_count_postgres_sink FROM target_count WITH (
    connector = 'jdbc',
    jdbc.url = 'jdbc:postgresql://postgres:5432/mydb',
    user = 'myuser',
    password = '123456',
    table.name = 'target_count',
    type = 'upsert',
    primary_key = 'target_id'
);

​

Verify update

To ensure that the target table has been updated, query from target_count in PostgreSQL.

SELECT * FROM target_count
LIMIT 10;

​

Data type mapping

For the PostgreSQL data type mapping table, see the Data type mapping table under the Ingest data from PostgreSQL CDC topic.

Additional notes regarding sinking data to PostgreSQL:

• A varchar column in RisingWave can be sinked to a uuid column in Postgres.
• Only one-dimensional arrays in RisingWave can be sinked to PostgreSQL.
• For array type, we only support smallint, integer, bigint, real, double precision, and varchar type now.