Flink这个问题怎么解决 ？

Flink这个问题怎么解决？--**--
-- Author: chenshengpeng
-- Created Time: 2024-03-11 16:04:01
-- Description: Write your description here
-- Hints: You can use SET statements to modify the configuration

-- Author: chenshengpeng
-- Created Time: 2024-02-29 18:21:32
-- Description: Write your description here
-- Hints: You can use SET statements to modify the configuration

--创建一个print_table临时表。

CREATE TEMPORARY TABLE print(
user_id bigint ,
member_id bigint,
user_name STRING,
phone STRING,
reg_time TIMESTAMP,
reg_type int,
client STRING,
user_type int,
user_level int,
idcard_no STRING,
idcard_name STRING,
idcard_address STRING,
vip_expire_time TIMESTAMP,
sm_device_id STRING,
system_version STRING,
register_ip STRING,
bind_phone STRING,
last_login_ip STRING,
last_login_time TIMESTAMP,
PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
'connector' = 'selectdb',
'load-url' = 'selectdb-cn-jia3ma3b003.selectdbfe.rds.aliyuncs.com:8080',
'jdbc-url' = 'selectdb-cn-jia3ma3b003.selectdbfe.rds.aliyuncs.com:9030',
'cluster-name' = 'new_cluster',
'table.identifier' = 'center_risk_service.data_model_user',
'username' = '',
'password' = '',
'sink.parallelism' = '3'
);

BEGIN STATEMENT SET;

-- INSERT INTO select-db.center_risk_service.data_model_user
INSERT INTO print
(
user_id
,member_id
,user_name
,phone
,reg_time
,reg_type
,client
,user_type
,user_level
,idcard_no
,idcard_name
,idcard_address
,vip_expire_time
,sm_device_id
,system_version
,register_ip
,bind_phone
,last_login_ip
,last_login_time
)
select
CAST(ui.user_id as bigint) user_id
,CAST(ui.member_id as bigint) member_id
,ui.user_name
,ui.phone
,ui.create_time as reg_time
,ui.reg_type
,ui.client
,CAST(ui.user_type as int) user_type
,ugc.level as user_level
,pua.idcard_no
,pua.idcard_name
,pua.idcard_address
,vi.expire_time vip_expire_time
,aall.device_id sm_device_id
,aall.system_version
,ma.register_ip
,ma.phone bind_phone
,ma.last_login_ip
,aall.login_time as last_login_time
from msyql.t_user_service.user_info ui