Oracle

Background

The Oracle connector allows you to read from and write to Oracle databases in streaming and batch modes using Ververica Platform. The connector is a direct extension of the JDBC Connector, sharing some of the capabilities and configuration options. The following table describes the connector's capabilities.

Prerequisites

• Access to an active Oracle database instance running at a known network address (host and port).
• The required tables must already exist in the target Oracle database.
• Valid credentials: a username and password with sufficient privileges to read from or write to the tables.
• Knowledge of the Oracle service name associated with your database instance (needed for the JDBC URL and catalog configuration).

Supported Versions

• Oracle 19c, 21c and 23c
• Oracle XE for local testing

Oracle JBDC URL Format

Database names maps to Oracle Service names. As a consequence, the JDBC URL should have the service name after /.
In particular, both catalog and connector assume the following URL format, with a double slash (//) after the @ and a single slash (/) before the service name.

1    jdbc:oracle:thin:@//<host>:<port>/<service_name>

• host: The hostname or IP address of your Oracle server.
• port: The port number (default is 1521).
• service_name: The Oracle service name (e.g., ORCL).

In some Oracle versions, the double slash after @ is optional but using it is recommended for compatibility. Always ensure your URL matches your Oracle server configuration and uses the correct service name.

Table Management

Limitations

Example: Defining Source and Sink Tables

You have two tables in your Oracle database:

• HR.ORDERS (source table)
• HR.ORDERS_ARCHIVE (sink table)

1. Define the source table in Flink.

1  CREATE TABLE oracle_orders (
2   id INT,
3   order_date TIMESTAMP,
4   amount DECIMAL(10, 2),
5   PRIMARY KEY (id) NOT ENFORCED
6) WITH (
7  'connector' = 'oracle',
8  'jdbc-url' = 'jdbc:oracle:thin:@//oracle.example.com:1521/ORCL',
9  'username' = 'flink_user',
10  'password' = 'flink_pw',
11  'table-name' = 'HR.ORDERS'
12);

1. Define the sink table in Flink.

1  CREATE TABLE oracle_orders_archive (
2   id INT,
3   order_date TIMESTAMP,
4   amount DECIMAL(10, 2),
5   PRIMARY KEY (id) NOT ENFORCED
6 ) WITH (
7   'connector' = 'oracle',
8   'jdbc-url' = 'jdbc:oracle:thin:@//oracle.example.com:1521/ORCL',
9   'username' = 'flink_user',
10   'password' = 'flink_pw',
11   'table-name' = 'HR.ORDERS_ARCHIVE'
12);
13

1. Insert data from source to sink.

1INSERT INTO oracle_orders_archive
2SELECT * FROM oracle_orders;
3

Properties

In the table below is a summary of supported connector options. These are typically set in the Flink SQL WITH clause.