"Flink CDC中 为什么这么写编译不过？

"Flink CDC中 为什么这么写编译不过？

CREATE TABLE RMS_STORAGE_OUT_CARDS(
C_ID INT,
C_STORAGE_OUT_ITEM_ID INT,
C_CARD_NUMBER STRING,
C_GOODS_ID INT,
C_STATUS INT,
C_CREATEDBY STRING,
C_CREATEDON TIMESTAMP,
C_CHANNEL_CODE STRING,
C_HOSPID STRING,
proc_time AS PROCTIME(),
PRIMARY KEY (C_ID) NOT ENFORCED
) WITH (
'connector' = 'oracle-cdc',
'hostname' = 'oracle12cdb.xxxx.com',
'port' = '1521',
'username' = 'misnew',
'password' = 'mis32006',
'database-name' = 'XXXXRAC',
'schema-name' = 'RMS',
'debezium.log.mining.strategy' = 'online_catalog',
'debezium.log.mining.continuous.mine' = 'true',
'debezium.include.schema.changes' = 'false',
'scan.incremental.snapshot.chunk.key-column' = 'C_ID',
'scan.incremental.snapshot.chunk.size' = '80960' ,
'scan.snapshot.fetch.size' = '1024',
'table-name' = 'RMS_STORAGE_OUT_CARDS'
);

CREATE TABLE RMS_GOODS_BATCH(
C_ID INT,
C_NAME STRING,
C_OUTER_ID INT,
C_TYPE INT,
C_CARDTYPE INT,
C_START_DATE TIMESTAMP,
C_END_DATE TIMESTAMP,
C_PRICE DECIMAL(10,2),
C_GOODS_ID INT,
C_ENABLE INT,
C_FLAG INT,
C_SMS_TPL STRING,
C_CREATEDBY STRING,
C_CREATEDON TIMESTAMP,
C_OP_USER STRING,
C_OP_TIME TIMESTAMP,
PRIMARY KEY (C_ID) NOT ENFORCED
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:oracle:thin:@oracle12cdb.xxxx.com:1521:xxxxrac',
'driver' = 'oracle.jdbc.driver.OracleDriver',
'username' = 'misnew',
'password' = 'mis32006',
'table-name' = 'RMS.RMS_GOODS_BATCH'
);

CREATE TABLE RMS_STORAGE_OUT_ITEM(
C_ID INT,
C_STORAGE_OUT_ID INT,
C_SALE_ITEM_ID INT,
C_BATCH_ID STRING,
C_GOODS_ID INT,
C_GOODS_CODE STRING,
C_SALE_PRICE DECIMAL(10,2),
C_CARD_TYPE STRING,
C_GOODS_NAME STRING,
C_AMOUNT INT,
C_REMARK STRING,
C_PRICE DECIMAL(10,2),
C_CHARGE_TYPE INT,
C_CHARGE_ID INT,
PRIMARY KEY (C_ID) NOT ENFORCED
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:oracle:thin:@oracle12cdb.xxxx.com:1521:xxxxrac',
'driver' = 'oracle.jdbc.driver.OracleDriver',
'username' = 'misnew',
'password' = 'mis32006',
'table-name' = 'RMS.RMS_STORAGE_OUT_ITEM'
);

CREATE TABLE sink_rms_storage_out_cards (
C_ID INT,
C_CARD_NUMBER STRING,
C_STATUS INT,
C_CREATEDBY STRING,
C_CREATEDON TIMESTAMP,
C_CHANNEL_CODE STRING,
C_HOSPID STRING,
CARD_TYPE INT,
SALE_PRICE DECIMAL(10,2),
C_CHARGE_TYPE INT,
PRIMARY KEY (C_ID) NOT ENFORCED
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://10.108.37.132:3306/test',
'driver' = 'com.mysql.cj.jdbc.Driver',
'username' = 'root',
'password' = '123456',
'table-name' = 'rms_storage_out_cards2');

insert into sink_rms_storage_out_cards
select
o.C_ID ,
o.C_CARD_NUMBER ,
o.C_STATUS ,
o.C_CREATEDBY ,
o.C_CREATEDO
"