notebook/996/flinksql/sqlcopy/hongbao_test.sql

--********************************************************************--
-- Author:         machaofan@1253758386586182.onaliyun.com
-- Created Time:   2023-09-07 11:06:37
-- Description:    Write your description here
-- Hints:          You can use SET statements to modify the configuration
--********************************************************************--

CREATE TEMPORARY TABLE ods_box_hongbao_log_pd(
        d VARCHAR
        ,i VARCHAR
        ,t  BIGINT
      --  ,eventTime as TO_TIMESTAMP( CAST(t/1000 as bigint),3)
        ,eventTime as TO_TIMESTAMP(t,3)
        ,WATERMARK FOR eventTime AS eventTime - INTERVAL '30' SECONDS  -- 定义事件时间，允许的最大窗口延迟为5s
) WITH (
    'connector' ='sls',
    'endPoint' ='cn-shanghai.log.aliyuncs.com',
    'accessId' ='LTAI5tEqx7wh38NsaFW7BJFN',
    'accessKey' ='cENYqeXBbFkVZeShFyPgTlIt1ouMpa',
    'startTime' ='2023-09-07 00:00:00',
    'project' ='996box-dev',
    'logStore' ='collect-dev',
    'consumerGroup' ='hongbaotest02'
);


-- CREATE  TEMPORARY TABLE ads_box_red_ad_statis_pd (
--  red_id bigint,
--  user_id bigint,
--  red_show_num bigint,
--  red_click_num bigint,
--  red_into_users bigint,
--  gmt_create   TIMESTAMP,
--  gmt_modify   TIMESTAMP,
--  ds VARCHAR,
-- PRIMARY KEY (red_id, user_id,ds) NOT ENFORCED
-- ) WITH (
-- --    'connector' = 'print',
-- --    'logger'='true'
--   'connector'='hologres',
--   'dbname'='sh996box_dev',
--   'tablename'='ads_box_red_ad_statis_pd',
--   'endpoint' = 'hgpostcn-cn-7mz2qe6e7002-cn-hangzhou-vpc.hologres.aliyuncs.com:80',
--   'username' = '${secret_values.996engine_ak}',
--   'password' = '${secret_values.996engine_sk}',
--   'mutatetype'='insertorignore'
-- );


CREATE  TEMPORARY TABLE ads_box_test (
 red_id bigint,
 user_id bigint,
 red_show_num bigint,
 red_click_num bigint,
 red_into_users bigint,
 gmt_create   TIMESTAMP,
 gmt_modify   TIMESTAMP,
 ds VARCHAR,
PRIMARY KEY (red_id, user_id,ds) NOT ENFORCED
) WITH (
--    'connector' = 'print',
--    'logger'='true'
  'connector'='hologres',
  'dbname'='sh996box_dev',
  'tablename'='ads_box_test',
  'endpoint' = 'hgpostcn-cn-7mz2qe6e7002-cn-hangzhou-vpc.hologres.aliyuncs.com:80',
  'username' = '${secret_values.996engine_ak}',
  'password' = '${secret_values.996engine_sk}',
  'mutatetype'='insertorignore'
);


create  TEMPORARY view  ads_box_red_ad_statis_pd1 as
select
        i
        ,eventTime
        ,cast(JSON_VALUE(d,'$.uid') as BIGINT) as uid
        ,cast(JSON_VALUE(d,'$.object_id') AS BIGINT)  as send_id
        ,cast(JSON_VALUE(d,'$.red_id') AS BIGINT) as red_id
from  ods_box_hongbao_log_pd where i in ('hongbao_1300','hongbao_1301','hongbao_1302');

-- -- 每分钟更新一次从零点开始截止到当前时刻的累计销售额
INSERT INTO ads_box_red_ad_statis_pd
SELECT
    red_id,
    send_id,
    COALESCE(max(case WHEN i='hongbao_1300' THEN  COALESCE(red_show_num,0) end ),0) as red_show_num,
    COALESCE(max(case WHEN i='hongbao_1301' THEN COALESCE(red_click_num,0) end ),0)  as red_click_num,
    COALESCE(max(case WHEN i='hongbao_1302' THEN COALESCE(red_click_num,0) end ),0)  as red_into_users,
    window_start,
    window_end ,
    SUBSTR(window_start,1,19) as ds
FROM
(SELECT
    window_start,
    window_end,
    send_id,
    i,
    red_id,
    count(uid) as red_show_num,
    count(distinct uid) as red_click_num -- 当天累计销售额
FROM TABLE(
    -- 定义窗口最大长度为一天的累计窗口，窗口滚动步长为1分钟
    CUMULATE(
        TABLE ads_box_red_ad_statis_pd1,
        DESCRIPTOR(eventTime),
        INTERVAL '1' MINUTES,
        INTERVAL '1' DAY))
GROUP BY window_start, window_end,send_id,i,red_id) as a1
GROUP BY  SUBSTR(window_start,1,19) ,window_start, window_end,send_id,red_id
;