> ## Documentation Index > Fetch the complete documentation index at: https://docs.risingwave.com/llms.txt > Use this file to discover all available pages before exploring further. # 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](https://github.com/risingwavelabs/risingwave/tree/main/integration%5Ftests) of the RisingWave repository. ## Set up a PostgreSQL database ### Set up a PostgreSQL RDS instance on AWS Here we will use a standard class instance without Multi-AZ deployment as an example. 1. Log in to the AWS console. Search “RDS” in services and select the RDS panel. 2. Create a database with **PostgreSQL** as the **Engine** type. We recommend setting up a username and password or using other security options. 3. When the new instance becomes available, click on its panel. 4. From the Connectivity panel, we can find the endpoint and connection port information. ### Connect to the RDS instance from Postgres Now we can connect to the RDS instance. Make sure you have installed psql on your local machine, and start a psql prompt. Fill in the endpoint, the port, and login credentials in the connection parameters. ```bash theme={null} psql --host = pg-to-rw.xxxxxx.us-east-1.rds.amazonaws.com --port=5432 --username=awsuser --password ``` For more login options, refer to the [RDS connection guide](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER%5FConnectToPostgreSQLInstance.html). ### Launch and set up PostgreSQL To install PostgreSQL locally, see their [download options](https://www.postgresql.org/download/). If you are using the demo version, connect to PostgreSQL with the following command. Ensure that all other programs are disconnected from port 5432. ```bash theme={null} psql postgresql://myuser:123456@127.0.0.1:5432/mydb ``` Ensure that the Postgres user is granted the following privileges on the used table with the following SQL query. ```sql theme={null} 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. ```sql theme={null} 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](/get-started/quickstart) 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](https://openjdk.org/projects/jdk/11/) or later versions installed in your environment. ## Create a sink ### Syntax ```sql theme={null} CREATE SINK [ IF NOT EXISTS ] sink_name [FROM sink_from | AS select_query] WITH ( connector = 'jdbc' | 'postgres', field_name = 'field', ... ); ``` ### Parameters (JDBC) | Parameter or clause | Description | | :--------------------- | :--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | sink\_name | Name of the sink to be created. | | sink\_from | A clause that specifies the direct source from which data will be output. `sink_from` can be a materialized view or a table. Either this clause or a SELECT query must be specified. | | AS select\_query | A SELECT query that specifies the data to be output to the sink. Either this query or a FROM clause must be specified. See [SELECT](/sql/commands/sql-select) for the syntax and examples of the SELECT command. | | connector | Sink connector should be `jdbc`. To switch from `jdbc` to `postgres`, set `stream_switch_jdbc_pg_to_native = true` under `[streaming.developer]`. | | jdbc.url | **Required**. The JDBC URL of the destination database necessary for the driver to recognize and connect to the database. | | user | The user name for the database connection. | | password | The password for the database connection. | | jdbc.query.timeout | Specifies the timeout for the operations to downstream. If not set, the default is 60s. | | jdbc.auto.commit | Controls whether to automatically commit transactions for JDBC sink. If not set, the default is false. | | table.name | **Required**. The table in the destination database you want to sink to. | | schema.name | The schema in the destination database you want to sink to. The default value is public. | | type | Sink data type. Supported types: | | primary\_key | Required if type is upsert. The primary key of the sink, which should match the primary key of the downstream table. | | tcp.keepalive.enable | Enables TCP keepalive for the connection to the downstream database. If not set, the default is `false`. | | tcp.keepalive.idle | The time (in seconds) the connection must be idle before the first keepalive probe is sent. Takes effect only when `tcp.keepalive.enable` is `true`. If not set, the default is `600` (10 minutes). | | tcp.keepalive.interval | The time (in seconds) between consecutive keepalive probes. Takes effect only when `tcp.keepalive.enable` is `true`. If not set, the default is `10`. | | tcp.keepalive.count | The number of unacknowledged keepalive probes before the connection is considered dead. Takes effect only when `tcp.keepalive.enable` is `true`. If not set, the default is `3`. | ### Parameters (Postgres Native) RisingWave introduced the native Postgres sink connector in v2.2.0, and the JDBC sink connector for Postgres will be deprecated in a future release. Existing JDBC sinks will be maintained in a backwards-compatible way. There are some [known issues](https://github.com/risingwavelabs/risingwave/issues/24183) for native Postgres Sink connector. If you encounter these, we recommend using the JDBC sink first. You can try switching your JDBC sinks to the native Postgres sink in place by setting `stream_switch_jdbc_pg_to_native = true` under `[streaming.developer]`. | Parameter or clause | Description | | :--------------------- | :-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | sink\_name | Name of the sink to be created. | | sink\_from | A clause that specifies the direct source from which data will be output. `sink_from` can be a materialized view or a table. Either this clause or a SELECT query must be specified. | | AS select\_query | A SELECT query that specifies the data to be output to the sink. Either this query or a FROM clause must be specified. See [SELECT](/sql/commands/sql-select) for the syntax and examples of the SELECT command. | | connector | Sink connector must be `postgres`. | | user | The user name for the database connection. | | password | The password for the database connection. | | database | **Required**. The database in the destination database you want to sink to. | | table | **Required**. The table in the destination database you want to sink to. | | schema | The schema in the destination database you want to sink to. The default value is public. | | type | Sink data type. Supported types: | | primary\_key | Required if type is upsert. The primary key of the sink, which should match the primary key of the downstream table. | | ssl\_mode | The `ssl.mode` parameter determines the level of SSL/TLS encryption for secure communication with Postgres. Accepted values are `disabled`, `preferred`, `required`, `verify-ca`, and `verify-full`. The default value is `disabled`. | | ssl\_root\_cert | Specify the root certificate secret. You must [create secret](/operate/manage-secrets) first and then use it here. | | tcp.keepalive.enable | Enables TCP keepalive for the connection to the downstream database. If not set, the default is `false`. | | tcp.keepalive.idle | The time (in seconds) the connection must be idle before the first keepalive probe is sent. Takes effect only when `tcp.keepalive.enable` is `true`. If not set, the default is `600` (10 minutes). | | tcp.keepalive.interval | The time (in seconds) between consecutive keepalive probes. Takes effect only when `tcp.keepalive.enable` is `true`. If not set, the default is `10`. | | tcp.keepalive.count | The number of unacknowledged keepalive probes before the connection is considered dead. Takes effect only when `tcp.keepalive.enable` is `true`. If not set, the default is `3`. | ## 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. ```sql theme={null} 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. ```sql theme={null} 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](/sql/commands/sql-create-sink) for more details. ```sql theme={null} 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. ```sql theme={null} SELECT * FROM target_count LIMIT 10; ``` ## Data type mapping For the PostgreSQL data type mapping table, see the [Data type mapping table](/integrations/sources/postgresql-cdc#data-type-mapping) 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.