Flink CDC中oracle 19c 环境版本下 oracle cdc 在flink sql？

在 Flink CDC 中，您可以使用 Flink SQL 来配置和使用 Oracle CDC Connector 来处理 Oracle 19c 环境下的 CDC 数据。以下是一个使用 Flink SQL 配置 Oracle CDC Connector 的示例：

CREATE TABLE my_table (
  id INT,
  name VARCHAR,
  age INT,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'oracle-cdc',
  'hostname' = 'your_hostname',
  'port' = '1521',
  'database-name' = 'your_database_name',
  'table-name' = 'your_table_name',
  'username' = 'your_username',
  'password' = 'your_password',
  'startup-mode' = 'earliest',
  'debezium.snapshot.mode' = 'initial',
  'debezium.snapshot.locking.mode' = 'none',
  'debezium.transforms' = 'unwrap',
  'debezium.transforms.unwrap.type' = 'io.debezium.transforms.ExtractNewRecordState',
  'debezium.transforms.unwrap.drop.tombstones' = 'false',
  'debezium.transforms.unwrap.delete.handling.mode' = 'rewrite',
  'debezium.transforms.unwrap.add.fields' = 'op,ts_ms',
  'debezium.transforms.unwrap.add.headers' = 'source',
  'debezium.transforms.unwrap.add.source.fields' = 'db,table',
  'debezium.transforms.unwrap.add.source.schema' = 'false',
  'debezium.transforms.unwrap.add.source.table' = 'false',
  'debezium.transforms.unwrap.add.source.timestamp' = 'false',
  'debezium.transforms.unwrap.add.source.version' = 'false'
)

在上述示例中，使用 'connector' = 'oracle-cdc' 指定了使用 Oracle CDC Connector。此外，还使用了 Debezium 库中提供的一些转换器（transforms），例如 io.debezium.transforms.ExtractNewRecordState 用于解包 CDC 数据，op 字段用于标识 CDC 操作类型等。

您可以根据实际情况修改示例中的连接信息和配置选项，以适应您的 Oracle 19c 环境和表结构。

link CDC 支持 Oracle 19c 环境下的 Oracle CDC，可以使用 Flink SQL 来配置和使用 Oracle CDC Connector。
以下是一个使用 Flink SQL 配置 Oracle CDC Connector 的示例：
sql
Copy
CREATE TABLE my_table (
id INT,
name VARCHAR,
age INT,
PRIMARY KEY (id) NOT ENFORCED
) WITH (
'connector' = 'oracle-cdc',
'hostname' = 'your_hostname',
'port' = '1521',
'database-name' = 'your_database_name',
'table-name' = 'your_table_name',
'username' = 'your_username',
'password' = 'your_password',
'startup-mode' = 'earliest',
'debezium.snapshot.mode' = 'initial',
'debezium.snapshot.locking.mode' = 'none',
'debezium.transforms' = 'unwrap',
'debezium.transforms.unwrap.type' = 'io.debezium.transforms.ExtractNewRecordState',
'debezium.transforms.unwrap.drop.tombstones' = 'false',
'debezium.transforms.unwrap.delete.handling.mode' = 'rewrite',
'debezium.transforms.unwrap.add.fields' = 'op,ts_ms',
'debezium.transforms.unwrap.add.headers' = 'source',
'debezium.transforms.unwrap.add.source.fields' = 'db,table',
'debezium.transforms.unwrap.add.source.schema' = 'false',
'debezium.transforms.unwrap.add.source.table' = 'false',
'debezium.transforms.unwrap.add.source.timestamp' = 'false',
'debezium.transforms.unwrap.add.source.version' = 'false'
)
在以上示例中，通过 connector 属性指定了使用 Oracle CDC Connector，同时使用了 Debezium 库中提供的一些转换器，例如 io.debezium.transforms.ExtractNewRecordState 用于解包 CDC 数据、op 字段用于标识 CDC 操作类型等。