AppendStreamTableSink不支持使用节点Join产生的更新更改（joinType=[InnerJoin]

Question

当我使用Flink SQL执行以下语句时，错误报告如下:

请求

根据< code>user_id字段对user_behavior_kafka_table中的数据进行分组，然后取出每组中< code>ts字段值最大的数据

Excute sql

SELECT user_id,item_id,ts FROM user_behavior_kafka_table AS a 
WHERE ts = (select max(b.ts) 
FROM user_behavior_kafka_table AS b 
WHERE a.user_id = b.user_id );

Flink 版本

1.11.2

错误消息

AppendStreamTableSink doesn't support consuming update changes which is produced by node Join(joinType=[InnerJoin], where=[((user_id = user_id0) AND (ts = EXPR$0))], select=[user_id, item_id, ts, user_id0, EXPR$0], leftInputSpec=[NoUniqueKey], rightInputSpec=[JoinKeyContainsUniqueKey])

作业部署

在纱线上

表消息

< li >来自消费者kafka主题的user_behavior_kafka_table数据

{“user_id”：“aaa”，“item_id”：“11-222-333”，“comment”：“aaa access item at”，“ts”：100}

{"user_id"："ccc"，"item_id"："11-222-334"，"注释"："ccc访问项at"，"ts"： 200}

｛“user_id”：“ccc”，“item_id:”11-222-333，“comment:”ccc访问项在“，”ts:”300｝

{“user_id”：“bbb”，“item_id”：“11-222-334”，“comment”：“bbb access item at”，“ts”：200}

｛“user_id”：“aaa”，“item_id:”11-222-333，“comment:”aaa访问项在“，”ts:”200｝

{user_id：aaa，item_id：11-222-334，注释：aaa访问项at，ts：400}

{"user_id"："ccc"，"item_id"："11-222-333"，"注释"："ccc访问项at"，"ts"： 400}

{user_id：vvv，item_id：11-222-334，注释：vvv访问项at，ts：200}

｛“user_id”：“bbb”，“item_id:”11-222-333，“comment:”bbb访问项在“，”ts:”300｝

{user_id：aaa，item_id：11-222-334，注释：aaa访问项at，ts：300}

｛“user_id”：“ccc”，“item_id:”11-222-333，“comment:”ccc访问项在“，”ts:”100｝

｛“user_id”：“bbb”，“item_id:”11-222-334，“comment:”bbb访问项在“，”ts:”100｝

• user_behavior_hive_table 预期结果

{user_id：aaa，item_id：11-222-334，注释：aaa访问项at，ts：400}

｛“user_id”：“bbb”，“item_id:”11-222-333，“comment:”bbb访问项在“，”ts:”300｝

{"user_id"："ccc"，"item_id"："11-222-333"，"注释"："ccc访问项at"，"ts"： 400}

{user_id：vvv，item_id：11-222-334，注释：vvv访问项at，ts：200}

Answer 1

要从该查询中获得预期结果，需要以批处理模式执行该查询。作为一个流式查询，Flink SQL planner无法处理它，如果可以，它将生成一个结果流，其中每个user_id的最后结果将与预期结果匹配，但会有额外的中间结果。

例如，对于用户 aaa，将显示以下结果：

aaa 11-222-333 100
aaa 11-222-333 200
aaa 11-222-334 400

但是将跳过 ts=300 的行，因为它从来不是 ts 最大值的行。

如果要在流式处理模式下执行此操作，请尝试将其重新设置为 top-n 查询：

SELECT user_id, item_id, ts FROM
FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY ts DESC) AS row_num
  FROM user_behavior_kafka_table)
WHERE row_num = 1;

我相信这应该有效，但我不能轻易测试它。