flink sql oracle 类型映射

在Apache Flink中，当使用Flink SQL连接Oracle数据库时，需要将Oracle数据库中的数据类型映射为Flink SQL能够识别的类型。以下是一些常见的Oracle数据类型以及它们在Flink SQL中的映射关系：

请注意，Oracle的NUMBER类型通常映射为Flink的DECIMAL类型，而VARCHAR2, CHAR, CLOB映射为STRING，DATE和TIMESTAMP映射为TIMESTAMP(3)。

在创建表时，你需要指定列的类型，如下所示：

CREATE TABLE oracle_table (
  id DECIMAL(20, 0),
  name VARCHAR(255),
  birthday TIMESTAMP(3),
  content CLOB,
  data BLOB
) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:oracle:thin:@//hostname:port/serviceName',
  'table-name' = 'oracle_table',
  'username' = 'username',
  'password' = 'password'
);

定义了一个名为oracle_table的表，其中包含了Oracle数据库中的一些常见类型，并在Flink SQL中将它们映射为对应的类型。

环境准备

要通过 Flink SQL 将 Oracle 数据库中的数据表插入到 Iceberg 中，你需要遵循以下步骤：
确保你的环境中已经安装了以下组件：

• Apache Flink
• Oracle JDBC 驱动
• MySQL

创建 Flink 表映射 Oracle 数据

使用 Flink SQL DDL 语句创建一个临时表，映射到 Oracle 数据库中的数据表。

CREATE TEMPORARY TABLE oracle_table (
-- 定义列，与 Oracle 表结构对应
) WITH (
'connector' = 'jdbc',
'url' = '<your-oracle-jdbc-url>',
'table-name' = '<your-oracle-table-name>',
'username' = '<your-oracle-username>',
'password' = '<your-oracle-password>',
'scan.fetch-size' = '1000' -- 可选，调整 JDBC fetch size 以优化性能
);

创建 MySQL 表

使用 Flink SQL DDL 语句创建 MySQL 表，这个表将用于存储从 Oracle 读取的数据。
使用 Flink SQL CLI 根据需要同步的数据创建 MySQL 源表，用于同步底层数据库表的数据

CREATE TABLE nbaplayers (
player_id INT,
team_id INT,
player_name VARCHAR,
height FLOAT,
PRIMARY KEY (player_id) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'hostname' = '192.168.10.60',
'port' = '3306',
'username' = 'root',
'password' = 'Admin@123!',
'database-name' = 'bigdata',
'table-name' = 'players',
'server-time-zone' = 'Asia/Shanghai'
);

插入数据

使用 Flink SQL 的 INSERT INTO 语句将 Oracle 表中的数据插入到 MySQL 表中。

INSERT INTO mysql_table SELECT * FROM oracle_table;

这个查询会触发 Flink 读取 Oracle 表中的数据，并写入到 MySQL 表中。

注意事项：

• 确保 Oracle JDBC 驱动与你的 Oracle 数据库版本兼容。
• 根据你的网络条件和数据量大小，调整 JDBC 连接和 Flink 作业的配置，以优化性能。
• 监控 Flink 作业的执行情况，确保数据能够正确无误地插入到 MySQL 表中。
• 如果遇到任何问题，检查 Flink 和 MySQL 的日志以获取更多信息。

通过遵循上述步骤，你可以使用 Flink SQL 将 Oracle 数据库中的数据表插入到 MySQL 中。记得在生产环境中进行充分的测试和验证，以确保数据的准确性和系统的稳定性。