Oracle有些表没有物理主键,同步时传统做法是使用rowid,目前在尝试使用flinkcdc进行数据同步,从flink schema中获取不到rowid。
是最新版本不支持,还是需要设置参数?
“Devil组”引证GPT后的撰写:
例如,使用以下 SQL 语句在 Oracle 中创建一个人工主键:
ALTER TABLE your_table ADD (id NUMBER(10) NOT NULL);
CREATE SEQUENCE your_table_seq START WITH 1 INCREMENT BY 1;
UPDATE your_table SET id = your_table_seq.NEXTVAL;
ALTER TABLE your_table ADD CONSTRAINT your_table_pk PRIMARY KEY (id);
然后,在 Flink CDC 中,可以使用以下代码指定主键并连接到 Oracle 数据库:
String ddl = "CREATE TABLE your_table (id INT NOT NULL, col1 STRING, col2 INT, PRIMARY KEY (id)) WITH (...)";
String jdbcUrl = "jdbc:oracle:thin:@localhost:1521:ORCL";
String username = "your_username";
String password = "your_password";
JdbcTableSource jdbcTableSource = JdbcTableSource.builder()
.setDrivername("oracle.jdbc.OracleDriver")
.setDBUrl(jdbcUrl)
.setUsername(username)
.setPassword(password)
.setTableName("your_table")
.setPrimaryKey("id")
.build();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(streamExecutionEnvironment);
tableEnv.registerTableSource("your_table", jdbcTableSource);
Table resultTable = tableEnv.sqlQuery("SELECT * FROM your_table");
要替换 your_table、jdbcUrl、username 和 password 为表名、Oracle 数据库连接信息和您的用户名和密码。同时,需要在 JdbcTableSource 中使用 setPrimaryKey 方法来指定主键。在这个示例中,用 id 字段作为主键。最后,在 StreamTableEnvironment 中使用 registerTableSource 方法将 JDBC 表注册到 Flink 的表环境中,然后就可以使用 SQL 查询来查询 Oracle 中的数据了。