如何通过Flink-sql 实现3个kafka-topic的join以及后续的窗口聚合计算操作？

[wang guanglei <gl...@outlook.com>]

同行们，大家好，

请教一个问题，现在有3个kafka的topic：device  consumer order
想用Flink-sql计算出多个聚合指标，比如：过去12小时，每个deviceid下的订单量。
我是这么设计的：

  1.  先通过 create table   with(...='kafka') ... 注册出 table1 table2 table3  ，指定事件时间、water mark
  2.  进行3张表的关联：

create temporary view wide_table as (
  select       ***
  from       table1
  join       table2          on ...
  join       table3          on ...
 )

  1.  在wide-table上进行聚合计算：

select deviceId
     ,HOP_START(voc.timestamp, INTERVAL '5' SECOND , INTERVAL '10' SECOND)
     ,count(1) as cnt
from wide_table as voc
group by HOP(voc.timestamp, INTERVAL '5' SECOND , INTERVAL '10' SECOND)
       ,deviceId

任务开始运行之后，一直没有聚合计算的结果，通过 select * from wide_table 可以看到明细。

经过很长一段时间的搜索，发现table1/2/3中的time-attr的字段，在经过join之后，time-attr属性会被丢弃，进而无法在窗口中使用。

请问大家，像这种多个topic的join，然后再聚合操作，用Flink-sql实现的话，怎么做呢？还是我使用方式上有问题呢？

非常感谢！！！

Re: 如何通过Flink-sql 实现3个kafka-topic的join以及后续的窗口聚合计算操作？

[Caizhi Weng <ts...@gmail.com>]

Hi！

双流 join 操作确实会丢弃 time attribute 字段。如果其中一些表变化不大的话，可以考虑改成维表 join 的形式就能留下 time
attribute。如果每张表都很容易变化，可以考虑使用新引入的 event time temporal join[1]，肯定能满足需求。

[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.13/zh/docs/dev/table/sql/queries/joins/#event-time-temporal-join

wang guanglei <gl...@outlook.com> 于2021年8月10日周二 上午9:35写道：

> 同行们，大家好，
>
> 请教一个问题，现在有3个kafka的topic：device  consumer order
> 想用Flink-sql计算出多个聚合指标，比如：过去12小时，每个deviceid下的订单量。
> 我是这么设计的：
>
>   1.  先通过 create table   with(...='kafka') ... 注册出 table1 table2 table3
> ，指定事件时间、water mark
>   2.  进行3张表的关联：
>
> create temporary view wide_table as (
>   select       ***
>   from       table1
>   join       table2          on ...
>   join       table3          on ...
>  )
>
>   1.  在wide-table上进行聚合计算：
>
> select deviceId
>      ,HOP_START(voc.timestamp, INTERVAL '5' SECOND , INTERVAL '10' SECOND)
>      ,count(1) as cnt
> from wide_table as voc
> group by HOP(voc.timestamp, INTERVAL '5' SECOND , INTERVAL '10' SECOND)
>        ,deviceId
>
> 任务开始运行之后，一直没有聚合计算的结果，通过 select * from wide_table 可以看到明细。
>
>
> 经过很长一段时间的搜索，发现table1/2/3中的time-attr的字段，在经过join之后，time-attr属性会被丢弃，进而无法在窗口中使用。
>
> 请问大家，像这种多个topic的join，然后再聚合操作，用Flink-sql实现的话，怎么做呢？还是我使用方式上有问题呢？
>
> 非常感谢！！！
>