flinksql的primary key异常处理
·
flink在upsert操作的时候, 是需要指定主键的, 否则会出现 requires that Table has a full primary keys if it is updated.
针对该问题, 给出的解决如下:
Exception in thread "main" org.apache.flink.table.api.TableException: UpsertStreamTableSink requires that Table has a full primary keys if it is updated.
at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecLegacySink.translateToPlanInternal(StreamExecLegacySink.scala:93)
at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecLegacySink.translateToPlanInternal(StreamExecLegacySink.scala:48)
at org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:58)
at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecLegacySink.translateToPlan(StreamExecLegacySink.scala:48)
at org.apache.flink.table.planner.delegation.StreamPlanner$$anonfun$translateToPlan$1.apply(StreamPlanner.scala:67)
flink1.11 后, 官方给出的解决思路是:
create table t_sink_01 (
id varchar,
name1 varchar,
name2 varchar,
ts timestamp(3),
money varchar,
primary key(id) NOT ENFORCED '这个是官方给出'
)
with(
'connector.username' = '******'
,'connector.type' = 'jdbc'
,'connector.password' = '******'
,'connector.table' = 't_student'
,'connector.url' = 'jdbc:mysql://******:3306/******'
,'connector.write.flush.max-rows' = '1'
) ;
'说明:'
primary key(id) NOT ENFORCED '意思是声明id为主键, 但是不做强检验!'
但是发现还是报同样的异常.
还是使用flink1.11之前的解决思路: 即套一个子查询, 进行group by 来让flink来推测出主键.
INSERT INTO t_snoring_flink_01
select
id ,
tsa,
total_anti_snore_count,
total_anti_snore_during
from (
select
CONCAT_WS('|',DATE_FORMAT(winTime, 'yyyy-MM-dd HH:mm:00'),deviceName ) as id,
MAX(winTime) AS tsa,
MAX(CAST(totalAntiSnoreCount AS integer)) AS total_anti_snore_count,
MAX(CAST(totalAntiSnoreDuring AS BIGINT)) AS total_anti_snore_during
from v_1144_4
group by CONCAT_WS('|',DATE_FORMAT(winTime, 'yyyy-MM-dd HH:mm:00'),deviceName )
) tt ;
我在这里使用是字符串拼接后生成的主键.
更多推荐
所有评论(0)