sorry,????????????????????????group agg. ????????tableConfig.setIdleStateRetentionTime(Time.minutes(2),Time.minutes(7))??????????????????????????????????????????????.
------------------ ???????? ------------------ ??????: "Benchao Li"<[email protected]>; ????????: 2020??7??6??(??????) ????12:52 ??????: "user-zh"<[email protected]>; ????: Re: ??????FLINKSQL1.10????????????UV ????????SQL?????????????????????????????????????????? ??????????????????????state retention[1]?????????????????????????????????????? [1] https://ci.apache.org/projects/flink/flink-docs-master/dev/table/streaming/query_configuration.html#idle-state-retention-time x <[email protected]> ??2020??7??6?????? ????11:15?????? > ??????1.10.1??????sink????????????????window??????count > distinct??????????????????????????????????window??????count > distinct??????????????????????????????????????????????window????????????????group&nbsp;DATE_FORMAT(rowtm, > 'yyyy-MM-dd') ????sql?????????????????????????? > val rt_totaluv_view : Table = tabEnv.sqlQuery( > """ > SELECT MAX(DATE_FORMAT(rowtm, 'yyyy-MM-dd HH:mm:00')) > time_str,COUNT(DISTINCT userkey) uv > FROM source > GROUP BY DATE_FORMAT(rowtm, 'yyyy-MM-dd') > """) > tabEnv.createTemporaryView("rt_totaluv_view",rt_totaluv_view) > > val totaluvTmp = tabEnv.toRetractStream[(String,Long)](rt_totaluv_view) > .filter( line =&gt; line._1 == true ).map( line =&gt; line._2 ) > > val totaluvTabTmp = tabEnv.fromDataStream( totaluvTmp ) > > tabEnv.sqlUpdate( > s""" > INSERT INTO mysql_totaluv > SELECT _1,MAX(_2) > FROM $totaluvTabTmp > GROUP BY _1 > """) > ------------------&nbsp;????????&nbsp;------------------ > ??????:&nbsp;"Benchao Li"<[email protected]&gt;; > ????????:&nbsp;2020??7??3??(??????) ????9:47 > ??????:&nbsp;"user-zh"<[email protected]&gt;; > > ????:&nbsp;Re: ??????FLINKSQL1.10????????????UV > > > > ??????????????????????????????????????????[1]??????window??????count distinct?????????????? > ??????????1.11?????????? > > [1] https://issues.apache.org/jira/browse/FLINK-17942 > > x <[email protected]&gt; ??2020??7??3?????? ????4:34?????? > > &gt; ????????????????????????????????checkpoint?????????????????????????????????? > &gt; > &gt; > ????????tableConfig.setIdleStateRetentionTime(Time.minutes(2),Time.minutes(7)),??????????????????????key???????????????????????????? > &gt; > &gt; > &gt; > &gt; > &gt; ------------------&amp;nbsp;????????&amp;nbsp;------------------ > &gt; ??????:&amp;nbsp;"Jark Wu"<[email protected]&amp;gt;; > &gt; ????????:&amp;nbsp;2020??6??18??(??????) ????12:16 > &gt; ??????:&amp;nbsp;"user-zh"<[email protected]&amp;gt;; > &gt; > &gt; ????:&amp;nbsp;Re: ??????FLINKSQL1.10????????????UV > &gt; > &gt; > &gt; > &gt; ?????????????????????????????? > &gt; > &gt; On Thu, 18 Jun 2020 at 10:34, x <[email protected]&amp;gt; wrote: > &gt; > &gt; &amp;gt; ??????1.10??????????????????,??????????????????????????????? > &gt; &amp;gt; val resTmpTab: Table = tabEnv.sqlQuery( > &gt; &amp;gt;&amp;nbsp;&amp;nbsp; """ > &gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; SELECT > MAX(DATE_FORMAT(ts, 'yyyy-MM-dd > &gt; HH:mm:00')) > &gt; &amp;gt; time_str,COUNT(DISTINCT userkey) uv > &gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; FROM > user_behavior&amp;nbsp;&amp;nbsp;&amp;nbsp; GROUP BY > &gt; DATE_FORMAT(ts, 'yyyy-MM-dd')&amp;nbsp;&amp;nbsp;&amp;nbsp; """) > &gt; &amp;gt; > &gt; &amp;gt; val > resTmpStream=tabEnv.toRetractStream[(String,Long)](resTmpTab) > &gt; &amp;gt;&amp;nbsp;&amp;nbsp; > &gt; .filter(line=&amp;amp;gt;line._1==true).map(line=&amp;amp;gt;line._2) > &gt; &amp;gt; > &gt; &amp;gt; val res= tabEnv.fromDataStream(resTmpStream) > &gt; &amp;gt; tabEnv.sqlUpdate( > &gt; &amp;gt;&amp;nbsp;&amp;nbsp; s""" > &gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; INSERT INTO > rt_totaluv > &gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; SELECT _1,MAX(_2) > &gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; FROM $res > &gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; GROUP BY _1 > &gt; &amp;gt;&amp;nbsp;&amp;nbsp;&amp;nbsp;&amp;nbsp; """) > &gt; &amp;gt; > &gt; &amp;gt; > &gt; &amp;gt; > ------------------&amp;amp;nbsp;????????&amp;amp;nbsp;------------------ > &gt; &amp;gt; ??????:&amp;amp;nbsp;"Jark Wu"<[email protected]&amp;amp;gt;; > &gt; &amp;gt; ????????:&amp;amp;nbsp;2020??6??17??(??????) ????1:55 > &gt; &amp;gt; ??????:&amp;amp;nbsp;"user-zh"<[email protected] > &amp;amp;gt;; > &gt; &amp;gt; > &gt; &amp;gt; ????:&amp;amp;nbsp;Re: ??????FLINKSQL1.10????????????UV > &gt; &amp;gt; > &gt; &amp;gt; > &gt; &amp;gt; > &gt; &amp;gt; ?? Flink 1.11 ???????????????????? > &gt; &amp;gt; > &gt; &amp;gt; CREATE TABLE mysql ( > &gt; &amp;gt; &amp;amp;nbsp;&amp;amp;nbsp; time_str STRING, > &gt; &amp;gt; &amp;amp;nbsp;&amp;amp;nbsp; uv BIGINT, > &gt; &amp;gt; &amp;amp;nbsp;&amp;amp;nbsp; PRIMARY KEY (ts) NOT ENFORCED > &gt; &amp;gt; ) WITH ( > &gt; &amp;gt; &amp;amp;nbsp;&amp;amp;nbsp; 'connector' = 'jdbc', > &gt; &amp;gt; &amp;amp;nbsp;&amp;amp;nbsp; 'url' = > 'jdbc:mysql://localhost:3306/mydatabase', > &gt; &amp;gt; &amp;amp;nbsp;&amp;amp;nbsp; 'table-name' = 'myuv' > &gt; &amp;gt; ); > &gt; &amp;gt; > &gt; &amp;gt; INSERT INTO mysql > &gt; &amp;gt; SELECT MAX(DATE_FORMAT(ts, 'yyyy-MM-dd HH:mm:00')), > &gt; COUNT(DISTINCT&amp;amp;nbsp; > &gt; &amp;gt; user_id) > &gt; &amp;gt; FROM user_behavior; > &gt; &amp;gt; > &gt; &amp;gt; On Wed, 17 Jun 2020 at 13:49, x <[email protected]&amp;amp;gt; > wrote: > &gt; &amp;gt; > &gt; &amp;gt; &amp;amp;gt; ??????????????????????"??????"????????????????????????????????????????????UV?? > &gt; &amp;gt; &amp;amp;gt; sink?????????? > &gt; &amp;gt; &amp;amp;gt; tm uv > &gt; &amp;gt; &amp;amp;gt; 2020/06/17 13:46:00 10000 > &gt; &amp;gt; &amp;amp;gt; 2020/06/17 13:47:00 20000 > &gt; &amp;gt; &amp;amp;gt; 2020/06/17 13:48:00 30000 > &gt; &amp;gt; &amp;amp;gt; > &gt; &amp;gt; &amp;amp;gt; > &gt; &amp;gt; &amp;amp;gt; group by ?????????????????????? > &gt; &amp;gt; &amp;amp;gt; > &gt; &amp;gt; &amp;amp;gt; > &gt; &amp;gt; &amp;amp;gt; > &gt; > ------------------&amp;amp;amp;nbsp;????????&amp;amp;amp;nbsp;------------------ > &gt; &amp;gt; &amp;amp;gt; ??????:&amp;amp;amp;nbsp;"Benchao Li"< > [email protected] > &gt; &amp;amp;amp;gt;; > &gt; &amp;gt; &amp;amp;gt; ????????:&amp;amp;amp;nbsp;2020??6??17??(??????) ????11:46 > &gt; &amp;gt; &amp;amp;gt; ??????:&amp;amp;amp;nbsp;"user-zh"< > [email protected] > &gt; &amp;amp;amp;gt;; > &gt; &amp;gt; &amp;amp;gt; > &gt; &amp;gt; &amp;amp;gt; ????:&amp;amp;amp;nbsp;Re: ??????FLINKSQL1.10????????????UV > &gt; &amp;gt; &amp;amp;gt; > &gt; &amp;gt; &amp;amp;gt; > &gt; &amp;gt; &amp;amp;gt; > &gt; &amp;gt; &amp;amp;gt; Hi?? > &gt; &amp;gt; &amp;amp;gt; ?????????????????????????????? > &gt; &amp;gt; &amp;amp;gt; 1. ??????????group by + mini batch > &gt; &amp;gt; &amp;amp;gt; 2. window???? + fast emit > &gt; &amp;gt; &amp;amp;gt; > &gt; &amp;gt; &amp;amp;gt; ????#1??group > by????????????????????????????????????????????????DATE_FORMAT(rowtm, > &gt; 'yyyy-MM-dd')?? > &gt; &amp;gt; &amp;amp;gt; ??????????????????????????????state retention??????????????????????[1] > ????????mini > &gt; batch???????????? > &gt; &amp;gt; &amp;amp;gt; ??????[2] ???????? > &gt; &amp;gt; &amp;amp;gt; > &gt; &amp;gt; &amp;amp;gt; ????#2????????????????????????tumble???????????????????????????????????????????????????? > &gt; &amp;gt; &amp;amp;gt; fast > &gt; emit????????????????????experimental??feature???????????????????????????????????????????????????????????? > &gt; &amp;gt; &amp;amp;gt; table.exec.emit.early-fire.enabled = true > &gt; &amp;gt; &amp;amp;gt; table.exec.emit.early-fire.delay = 60 s > &gt; &amp;gt; &amp;amp;gt; > &gt; &amp;gt; &amp;amp;gt; [1] > &gt; &amp;gt; &amp;amp;gt; > &gt; &amp;gt; &amp;amp;gt; > &gt; &amp;gt; > &gt; > https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/table/streaming/query_configuration.html > &gt; &amp;gt; &amp;amp;gt; [2] > &gt; &amp;gt; &amp;amp;gt; > &gt; &amp;gt; &amp;amp;gt; > &gt; &amp;gt; > &gt; > https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/table/config.html > &gt; &amp;gt; &amp;amp;gt; > &gt; &amp;gt; &amp;amp;gt; x <[email protected]&amp;amp;amp;gt; > ??2020??6??17?????? ????11:14?????? > &gt; &amp;gt; &amp;amp;gt; > &gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; > &gt; ??????????????????????0??????????????UV??????????????????????????????????????UV?????????????????????? > &gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; CREATE VIEW uv_per_10min AS > &gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; SELECT&amp;amp;amp;amp;nbsp; > &gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; > &gt; MAX(DATE_FORMAT(proctime&amp;amp;amp;amp;nbsp;, > &gt; &amp;gt; 'yyyy-MM-dd > &gt; &amp;gt; &amp;amp;gt; HH:mm:00'))&amp;amp;amp;amp;nbsp;OVER w > &gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; AS > time_str,&amp;amp;amp;amp;nbsp; > &gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; &amp;amp;amp;amp;nbsp; > COUNT(DISTINCT user_id) OVER > &gt; w AS uv > &gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; FROM user_behavior > &gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; WINDOW w AS (ORDER BY proctime > ROWS BETWEEN > &gt; UNBOUNDED > &gt; &amp;gt; PRECEDING AND > &gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; CURRENT ROW); > &gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; > &gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; > &gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; ?????????????????????????? > &gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; PARTITION BY > DATE_FORMAT(rowtm, 'yyyy-MM-dd') > &gt; &amp;gt; ?????????????????????????????????? > &gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; PS??1.10??????????DDL??????????CREATE > VIEW?? > &gt; &amp;gt; &amp;amp;gt; &amp;amp;amp;gt; ???? > > > > -- > > Best, > Benchao Li -- Best, Benchao Li
