当前位置：网站首页>Flinksql multi table (three table) join/interval join

Flinksql multi table (three table) join/interval join

2022-07-27 01:06:00 【Me fan】

Go straight up sql And data flow diagram

Three tables flink_kafka_join_click, flink_kafka_join_pay, flink_kafka_join_price

All three tables are similar , The watchers mainly pay attention to the following join Part can be

create table flink_kafka_join_click (
    userid STRING,
    click STRING,
    ctime TIMESTAMP(3) ,
    procTime as PROCTIME (),
    -- ets AS TO_TIMESTAMP (FROM_UNIXTIME (ctime / 1000)),
    WATERMARK FOR ctime AS ctime - INTERVAL '5' MINUTE
  )
  with (
    'connector' = 'kafka',
    'topic' = 'data_flink_clickT',
    'properties.bootstrap.servers' = '',
    'properties.group.id' = 'data_flink_uat_click',
    'format' = 'json',
    'scan.startup.mode' = 'latest-offset'
  );

One ： join(regular join)

insert into `flinkThreeWaterT`
select
  ck.userid, ck.click, py.payway, pc.price, ck.ctime, py.ptime, pc.etime
from (
    select
      userid, click, TO_TIMESTAMP (ctime) as ctime
    from
      flink_kafka_join_click) ck left
      join (
        select
          userid, payway, TO_TIMESTAMP (ptime) as ptime
        from
          flink_kafka_join_pay
      ) py
      on ck.userid = py.userid left
      join (
        select
          userid, price, TO_TIMESTAMP (etime) as etime
        from
          flink_kafka_join_price
      ) pc
      on ck.userid = pc.userid;

here regular join Design to watermark The field of should be converted to timestamp

Two ： interval join

insert into interval_join_groupT
select
  DATE_FORMAT (ctime, 'yyyy-MM-dd') as day_time, userid, count (1) as num
from (
    select
      a.userid,
      a.click,
      b.payway,
      c.price,
      a.ctime,
      TO_TIMESTAMP (b.ptime) as ptime,
      TO_TIMESTAMP (c.etime) AS etime
    from (
        select
          userid, click, ctime
        from
          flink_kafka_join_click
        where
          click is not null
      ) a left
      join (
        select
          userid, payway, ptime
        from
          flink_kafka_join_pay
        where
          payway is not null
      ) b
      on a.userid = b.userid
      and a.ctime between b.ptime - INTERVAL '1' MINUTE
      and b.ptime + INTERVAL '1' MINUTE left
      join (
        select
          userid, price, etime
        from
          `flink_kafka_join_price`
        where
          price is not null
      ) c
      on a.userid = c.userid
      and a.ctime between c.etime - INTERVAL '1' MINUTE
      and c.etime + INTERVAL '1' MINUTE
  )
group by DATE_FORMAT (ctime, 'yyyy-MM-dd'), userid, payway;