This topic provides the DDL syntax that is used to create a Java Database Connectivity (JDBC) result table, describes the parameters in the WITH clause, and provides sample code.

Limits

Only Realtime Compute for Apache Flink that uses Ververica Runtime (VVR) 6.0.1 or later supports the JDBC connector.

Precautions

Fully managed Flink supports only the open source JDBC connector that does not include the JDBC driver for a specific database. When you use the JDBC connector, you must manually upload the JAR package of the driver for the database that you want to use. The following table describes the JDBC drivers supported by fully managed Flink.
Driver Group ID Artifact ID
MySQL mysql mysql-connector-java
Oracle com.oracle.database.jdbc ojdbc8
PostgreSQL org.postgresql postgresql
Note If you use a JDBC driver that is not listed in the table, you must test the correctness and availability of the JDBC driver.

DDL syntax

CREATE TABLE jdbc_sink (
  `id` BIGINT,
  `name` VARCHAR,
   PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:xxx',
  'table-name' = '<yourTable>',
  'username' = '<yourUsername>',
  'password' = '<yourPassword>'
);

Parameters in the WITH clause

Parameter Description Type Required Remarks
connector The type of the result table. STRING Yes Set the value to jdbc.
url The URL of the database. STRING Yes N/A.
table-name The name of the JDBC table. STRING Yes N/A.
username The name of the JDBC user. STRING No If you configure one of the username and password parameters, you must also configure the other parameter.
password The password of the JDBC user. STRING No N/A.
sink.buffer-flush.max-rows The maximum number of data records that can be cached before the flush operation is performed. INTEGER No Default value: 100. If you set this parameter to 0, data records are not cached before the flush operation is performed.
sink.buffer-flush.interval The interval at which data is flushed.

If data records are cached for a period that exceeds the duration specified by this parameter, the flush operation is performed in an asynchronous thread.

DURATION No Default value: 1. If you set this parameter to 0, data records are not cached before the flush operation is performed.
Note If you want to asynchronously process cached flush events, you can set the sink.buffer-flush.max-rows parameter to 0 and configure the sink.buffer-flush.interval parameter based on your business requirements.
sink.max-retries The maximum number of retries when data fails to be written to the database. INTEGER No Default value: 3.

Sample code

CREATE TEMPORARY TABLE datagen_source (
  `name` VARCHAR,
  `age` INT
) WITH (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE jdbc_sink (
  `name` VARCHAR,
  `age` INT
) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:xxxx',
  'table-name' = '<yourTable>',
  'username' = '<yourUsername>',
  'password' = '<yourPassword>'
);

INSERT INTO jdbc_sink
SELECT * FROM datagen_source;