You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user-zh@flink.apache.org by 赵一旦 <hi...@gmail.com> on 2020/12/15 14:48:06 UTC
FlinkSQL多个select group by统计union后输出，或多个select union后在group by统计的区别，以及问题。
需要，针对某个表，按照key1（xxx+yyy+ky1），key2（xxx+yyy+ky2），....等多组key统计。其中xxx+yyy为共同字段。目前有如下3种实现我。
（1）每组key分别统计，分别insert。
（2）每组key分别统计，然后union结果，然后insert。
（3）针对表多次select，然后union，然后再基于key统计，然后insert。
第三种方案中，会将ky1、ky2这几个不同的字段通过
select 'ky1' as key_name, ky1 as key_value
union
select 'ky2' as key_name, ky2 as key_value
的方式统一为key这个字段，最后通过（xxx+yyy+key_name+key_value）的方式统计。
目前发现个问题，方案3中，window结点一直没有watermark，导致不发生计算。
Re: FlinkSQL多个select group by统计union后输出，或多个select union后在group by统计的区别，以及问题。
Posted by 赵一旦 <hi...@gmail.com>.
为了方便描述，重新给出了完整SQL，以及部分分析到如下地址。
https://www.yuque.com/sixhours-gid0m/eegye3/xrz2mm
欢迎大家帮忙解答。
赵一旦 <hi...@gmail.com> 于2020年12月16日周三 上午10:52写道：
>
> 从这2个方案的source结点来看没有太大区别。但问题在于，我从web-ui的metric标签查看outputwatermark的时候。发现方案2中0号并行实例存在8个带有outputwatermark的指标（1个source开头，7个calc开头）。方案3中则只有2个。
>
> 赵一旦 <hi...@gmail.com> 于2020年12月16日周三 上午10:41写道：
>
>> 有没有人懂啊。今天的新发现如下。
>> 我看了下我的source结点的WEB-UI上展示的那个名字，然后在文本编辑器中划分了下。发现如下。
>> 方案2:
>>
>> Source: TableSourceScan(table=[[default_catalog, default_database, baidu_log, watermark=[-(TO_TIMESTAMP(FROM_UNIXTIME(/(CASE(IS NOT NULL($1), CAST($1):BIGINT NOT NULL, 0:BIGINT), 1000))), 60000:INTERVAL SECOND)]]], fields=[cid, server_time, d])
>> -> (
>>
>> Calc(select=[(d ITEM _UTF-16LE'106') AS $f0, ((d ITEM _UTF-16LE'77') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'77')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS $f1, Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time]),
>>
>> Calc(select=[(d ITEM _UTF-16LE'106') AS $f0, ((d ITEM _UTF-16LE'79') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'79')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS $f1, Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time]),
>>
>> Calc(select=[(d ITEM _UTF-16LE'106') AS $f0, ((d ITEM _UTF-16LE'80') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'80')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS $f1, Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time]),
>>
>> Calc(select=[(d ITEM _UTF-16LE'106') AS $f0, ((d ITEM _UTF-16LE'81') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'81')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS $f1, Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time]),
>>
>> Calc(select=[(d ITEM _UTF-16LE'106') AS $f0, ((d ITEM _UTF-16LE'83') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'83')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS $f1, Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time]),
>>
>> Calc(select=[(d ITEM _UTF-16LE'106') AS $f0, ((d ITEM _UTF-16LE'84') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'84')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS $f1, Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time]),
>>
>> Calc(select=[(d ITEM _UTF-16LE'106') AS $f0, ((d ITEM _UTF-16LE'86') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'86')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS $f1, Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time])
>> )
>>
>> 方案3:
>>
>> Source: TableSourceScan(table=[[default_catalog, default_database, dr1, watermark=[-(TO_TIMESTAMP(FROM_UNIXTIME(/(CASE(IS NOT NULL($1), CAST($1):BIGINT NOT NULL, 0:BIGINT), 1000))), 60000:INTERVAL SECOND)]]], fields=[cid, server_time, d])
>> -> (
>>
>> Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time, (d ITEM _UTF-16LE'106') AS su
>>
>> pply_id, _UTF-16LE'd107':VARCHAR(4) CHARACTER SET "UTF-16LE" AS field_key, ((d ITEM _UTF-16LE'107') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'107')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS field_value]),
>>
>>
>> Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time, (d ITEM _UTF-16LE'106') AS supply_id, _UTF-16LE'd77':VARCHAR(4) CHARACTER SET "UTF-16LE" AS field_key, ((d ITEM _UTF-16LE'77') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'77')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS field_value]),
>>
>>
>> Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time, (d ITEM _UTF-16LE'106') AS supply_id, _UTF-16LE'd79':VARCHAR(4) CHARACTER SET "UTF-16LE" AS field_key, ((d ITEM _UTF-16LE'79') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'79')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS field_value]),
>>
>>
>> Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time, (d ITEM _UTF-16LE'106') AS supply_id, _UTF-16LE'd80':VARCHAR(4) CHARACTER SET "UTF-16LE" AS field_key, ((d ITEM _UTF-16LE'80') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'80')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS field_value]),
>>
>>
>> Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time, (d ITEM _UTF-16LE'106') AS supply_id, _UTF-16LE'd81':VARCHAR(4) CHARACTER SET "UTF-16LE" AS field_key, ((d ITEM _UTF-16LE'81') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'81')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS field_value]),
>>
>>
>> Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time, (d ITEM _UTF-16LE'106') AS supply_id, _UTF-16LE'd83':VARCHAR(4) CHARACTER SET "UTF-16LE" AS field_key, ((d ITEM _UTF-16LE'83') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'83')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS field_value]),
>>
>>
>> Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time, (d ITEM _UTF-16LE'106') AS supply_id, _UTF-16LE'd84':VARCHAR(4) CHARACTER SET "UTF-16LE" AS field_key, ((d ITEM _UTF-16LE'84') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'84')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS field_value]),
>>
>>
>> Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time, (d ITEM _UTF-16LE'106') AS supply_id, _UTF-16LE'd86':VARCHAR(4) CHARACTER SET "UTF-16LE" AS field_key, ((d ITEM _UTF-16LE'86') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'86')) CASE _UTF-16LE'NULL':VARCHAR(214
>> 7483647) CHARACTER SET "UTF-16LE") AS field_value])
>> )
>>
>>
>> 赵一旦 <hi...@gmail.com> 于2020年12月15日周二 下午10:50写道：
>>
>>> 方案2没问题，方案3的window算子部分没有watermark。
>>>
>>> 赵一旦 <hi...@gmail.com> 于2020年12月15日周二 下午10:49写道：
>>>
>>>> 具体SQL如下。
>>>> 方案2：
>>>>
>>>>
>>>> INSERT INTO flink_sdk_stats
>>>> (
>>>> SELECT
>>>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>>>> sid AS `supply_id`,
>>>> 'd77' AS `field_key`,
>>>> d77 AS `filed_value`,
>>>> count(1) AS `pv`
>>>> FROM
>>>> baidu_log_view
>>>> GROUP BY
>>>> sid,
>>>> d77,
>>>> TUMBLE(event_time, INTERVAL '5' MINUTE)
>>>>
>>>> UNION ALL
>>>>
>>>> SELECT
>>>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>>>> sid AS `supply_id`,
>>>> 'd79' AS `field_key`,
>>>> d79 AS `filed_value`,
>>>> count(1) AS `pv`
>>>> FROM
>>>> baidu_log_view
>>>> GROUP BY
>>>> sid,
>>>> d79,
>>>> TUMBLE(event_time, INTERVAL '5' MINUTE)
>>>>
>>>> UNION ALL
>>>>
>>>> SELECT
>>>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>>>> sid AS `supply_id`,
>>>> 'd80' AS `field_key`,
>>>> d80 AS `filed_value`,
>>>> count(1) AS `pv`
>>>> FROM
>>>> baidu_log_view
>>>> GROUP BY
>>>> sid,
>>>> d80,
>>>> TUMBLE(event_time, INTERVAL '5' MINUTE)
>>>>
>>>> UNION ALL
>>>>
>>>> SELECT
>>>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>>>> sid AS `supply_id`,
>>>> 'd81' AS `field_key`,
>>>> d81 AS `filed_value`,
>>>> count(1) AS `pv`
>>>> FROM
>>>> baidu_log_view
>>>> GROUP BY
>>>> sid,
>>>> d81,
>>>> TUMBLE(event_time, INTERVAL '5' MINUTE)
>>>>
>>>> UNION ALL
>>>>
>>>> SELECT
>>>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>>>> sid AS `supply_id`,
>>>> 'd83' AS `field_key`,
>>>> d83 AS `filed_value`,
>>>> count(1) AS `pv`
>>>> FROM
>>>> baidu_log_view
>>>> GROUP BY
>>>> sid,
>>>> d83,
>>>> TUMBLE(event_time, INTERVAL '5' MINUTE)
>>>>
>>>> UNION ALL
>>>>
>>>> SELECT
>>>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>>>> sid AS `supply_id`,
>>>> 'd84' AS `field_key`,
>>>> d84 AS `filed_value`,
>>>> count(1) AS `pv`
>>>> FROM
>>>> baidu_log_view
>>>> GROUP BY
>>>> sid,
>>>> d84,
>>>> TUMBLE(event_time, INTERVAL '5' MINUTE)
>>>>
>>>> UNION ALL
>>>>
>>>> SELECT
>>>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>>>> sid AS `supply_id`,
>>>> 'd86' AS `field_key`,
>>>> d86 AS `field_value`,
>>>> count(1) AS `pv`
>>>> FROM
>>>> baidu_log_view
>>>> GROUP BY
>>>> sid,
>>>> d86,
>>>> TUMBLE(event_time, INTERVAL '5' MINUTE)
>>>> );
>>>>
>>>>
>>>>
>>>> 方案3:
>>>>
>>>>
>>>> INSERT INTO flink_sdk_stats
>>>> SELECT
>>>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>>>> `supply_id`,
>>>> `field_key`,
>>>> `field_value`,
>>>> count(1) AS `pv`
>>>> FROM
>>>> (
>>>> SELECT event_time, sid AS `supply_id`, 'd107' AS `field_key`, d107 AS `field_value` FROM baidu_log_view
>>>> UNION ALL
>>>> SELECT event_time, sid AS `supply_id`, 'd77' AS `field_key`, d77 AS `field_value` FROM baidu_log_view
>>>> UNION ALL
>>>> SELECT event_time, sid AS `supply_id`, 'd77' AS `field_key`, d77 AS `field_value` FROM baidu_log_view
>>>> UNION ALL
>>>> SELECT event_time, sid AS `supply_id`, 'd79' AS `field_key`, d79 AS `field_value` FROM baidu_log_view
>>>> UNION ALL
>>>> SELECT event_time, sid AS `supply_id`, 'd80' AS `field_key`, d80 AS `field_value` FROM baidu_log_view
>>>> UNION ALL
>>>> SELECT event_time, sid AS `supply_id`, 'd81' AS `field_key`, d81 AS `field_value` FROM baidu_log_view
>>>> UNION ALL
>>>> SELECT event_time, sid AS `supply_id`, 'd83' AS `field_key`, d83 AS `field_value` FROM baidu_log_view
>>>> UNION ALL
>>>> SELECT event_time, sid AS `supply_id`, 'd84' AS `field_key`, d84 AS `field_value` FROM baidu_log_view
>>>> UNION ALL
>>>> SELECT event_time, sid AS `supply_id`, 'd86' AS `field_key`, d86 AS `field_value` FROM baidu_log_view
>>>> )
>>>> GROUP BY
>>>> `supply_id`, `field_key`, `field_value`, TUMBLE(event_time, INTERVAL '5' MINUTE);
>>>>
>>>>
>>>> 赵一旦 <hi...@gmail.com> 于2020年12月15日周二 下午10:48写道：
>>>>
>>>>>
>>>>> 需要，针对某个表，按照key1（xxx+yyy+ky1），key2（xxx+yyy+ky2），....等多组key统计。其中xxx+yyy为共同字段。目前有如下3种实现我。
>>>>> （1）每组key分别统计，分别insert。
>>>>> （2）每组key分别统计，然后union结果，然后insert。
>>>>> （3）针对表多次select，然后union，然后再基于key统计，然后insert。
>>>>> 第三种方案中，会将ky1、ky2这几个不同的字段通过
>>>>>
>>>>> select 'ky1' as key_name, ky1 as key_value
>>>>> union
>>>>> select 'ky2' as key_name, ky2 as key_value
>>>>>
>>>>> 的方式统一为key这个字段，最后通过（xxx+yyy+key_name+key_value）的方式统计。
>>>>>
>>>>> 目前发现个问题，方案3中，window结点一直没有watermark，导致不发生计算。
>>>>>
>>>>>
>>>>>
>>>>>
Re: FlinkSQL多个select group by统计union后输出，或多个select union后在group by统计的区别，以及问题。
Posted by 赵一旦 <hi...@gmail.com>.
从这2个方案的source结点来看没有太大区别。但问题在于，我从web-ui的metric标签查看outputwatermark的时候。发现方案2中0号并行实例存在8个带有outputwatermark的指标（1个source开头，7个calc开头）。方案3中则只有2个。
赵一旦 <hi...@gmail.com> 于2020年12月16日周三 上午10:41写道：
> 有没有人懂啊。今天的新发现如下。
> 我看了下我的source结点的WEB-UI上展示的那个名字，然后在文本编辑器中划分了下。发现如下。
> 方案2:
>
> Source: TableSourceScan(table=[[default_catalog, default_database, baidu_log, watermark=[-(TO_TIMESTAMP(FROM_UNIXTIME(/(CASE(IS NOT NULL($1), CAST($1):BIGINT NOT NULL, 0:BIGINT), 1000))), 60000:INTERVAL SECOND)]]], fields=[cid, server_time, d])
> -> (
>
> Calc(select=[(d ITEM _UTF-16LE'106') AS $f0, ((d ITEM _UTF-16LE'77') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'77')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS $f1, Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time]),
>
> Calc(select=[(d ITEM _UTF-16LE'106') AS $f0, ((d ITEM _UTF-16LE'79') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'79')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS $f1, Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time]),
>
> Calc(select=[(d ITEM _UTF-16LE'106') AS $f0, ((d ITEM _UTF-16LE'80') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'80')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS $f1, Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time]),
>
> Calc(select=[(d ITEM _UTF-16LE'106') AS $f0, ((d ITEM _UTF-16LE'81') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'81')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS $f1, Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time]),
>
> Calc(select=[(d ITEM _UTF-16LE'106') AS $f0, ((d ITEM _UTF-16LE'83') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'83')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS $f1, Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time]),
>
> Calc(select=[(d ITEM _UTF-16LE'106') AS $f0, ((d ITEM _UTF-16LE'84') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'84')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS $f1, Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time]),
>
> Calc(select=[(d ITEM _UTF-16LE'106') AS $f0, ((d ITEM _UTF-16LE'86') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'86')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS $f1, Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time])
> )
>
> 方案3:
>
> Source: TableSourceScan(table=[[default_catalog, default_database, dr1, watermark=[-(TO_TIMESTAMP(FROM_UNIXTIME(/(CASE(IS NOT NULL($1), CAST($1):BIGINT NOT NULL, 0:BIGINT), 1000))), 60000:INTERVAL SECOND)]]], fields=[cid, server_time, d])
> -> (
>
> Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time, (d ITEM _UTF-16LE'106') AS su
>
> pply_id, _UTF-16LE'd107':VARCHAR(4) CHARACTER SET "UTF-16LE" AS field_key, ((d ITEM _UTF-16LE'107') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'107')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS field_value]),
>
>
> Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time, (d ITEM _UTF-16LE'106') AS supply_id, _UTF-16LE'd77':VARCHAR(4) CHARACTER SET "UTF-16LE" AS field_key, ((d ITEM _UTF-16LE'77') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'77')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS field_value]),
>
>
> Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time, (d ITEM _UTF-16LE'106') AS supply_id, _UTF-16LE'd79':VARCHAR(4) CHARACTER SET "UTF-16LE" AS field_key, ((d ITEM _UTF-16LE'79') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'79')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS field_value]),
>
>
> Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time, (d ITEM _UTF-16LE'106') AS supply_id, _UTF-16LE'd80':VARCHAR(4) CHARACTER SET "UTF-16LE" AS field_key, ((d ITEM _UTF-16LE'80') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'80')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS field_value]),
>
>
> Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time, (d ITEM _UTF-16LE'106') AS supply_id, _UTF-16LE'd81':VARCHAR(4) CHARACTER SET "UTF-16LE" AS field_key, ((d ITEM _UTF-16LE'81') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'81')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS field_value]),
>
>
> Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time, (d ITEM _UTF-16LE'106') AS supply_id, _UTF-16LE'd83':VARCHAR(4) CHARACTER SET "UTF-16LE" AS field_key, ((d ITEM _UTF-16LE'83') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'83')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS field_value]),
>
>
> Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time, (d ITEM _UTF-16LE'106') AS supply_id, _UTF-16LE'd84':VARCHAR(4) CHARACTER SET "UTF-16LE" AS field_key, ((d ITEM _UTF-16LE'84') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'84')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS field_value]),
>
>
> Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time, (d ITEM _UTF-16LE'106') AS supply_id, _UTF-16LE'd86':VARCHAR(4) CHARACTER SET "UTF-16LE" AS field_key, ((d ITEM _UTF-16LE'86') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'86')) CASE _UTF-16LE'NULL':VARCHAR(214
> 7483647) CHARACTER SET "UTF-16LE") AS field_value])
> )
>
>
> 赵一旦 <hi...@gmail.com> 于2020年12月15日周二 下午10:50写道：
>
>> 方案2没问题，方案3的window算子部分没有watermark。
>>
>> 赵一旦 <hi...@gmail.com> 于2020年12月15日周二 下午10:49写道：
>>
>>> 具体SQL如下。
>>> 方案2：
>>>
>>>
>>> INSERT INTO flink_sdk_stats
>>> (
>>> SELECT
>>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>>> sid AS `supply_id`,
>>> 'd77' AS `field_key`,
>>> d77 AS `filed_value`,
>>> count(1) AS `pv`
>>> FROM
>>> baidu_log_view
>>> GROUP BY
>>> sid,
>>> d77,
>>> TUMBLE(event_time, INTERVAL '5' MINUTE)
>>>
>>> UNION ALL
>>>
>>> SELECT
>>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>>> sid AS `supply_id`,
>>> 'd79' AS `field_key`,
>>> d79 AS `filed_value`,
>>> count(1) AS `pv`
>>> FROM
>>> baidu_log_view
>>> GROUP BY
>>> sid,
>>> d79,
>>> TUMBLE(event_time, INTERVAL '5' MINUTE)
>>>
>>> UNION ALL
>>>
>>> SELECT
>>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>>> sid AS `supply_id`,
>>> 'd80' AS `field_key`,
>>> d80 AS `filed_value`,
>>> count(1) AS `pv`
>>> FROM
>>> baidu_log_view
>>> GROUP BY
>>> sid,
>>> d80,
>>> TUMBLE(event_time, INTERVAL '5' MINUTE)
>>>
>>> UNION ALL
>>>
>>> SELECT
>>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>>> sid AS `supply_id`,
>>> 'd81' AS `field_key`,
>>> d81 AS `filed_value`,
>>> count(1) AS `pv`
>>> FROM
>>> baidu_log_view
>>> GROUP BY
>>> sid,
>>> d81,
>>> TUMBLE(event_time, INTERVAL '5' MINUTE)
>>>
>>> UNION ALL
>>>
>>> SELECT
>>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>>> sid AS `supply_id`,
>>> 'd83' AS `field_key`,
>>> d83 AS `filed_value`,
>>> count(1) AS `pv`
>>> FROM
>>> baidu_log_view
>>> GROUP BY
>>> sid,
>>> d83,
>>> TUMBLE(event_time, INTERVAL '5' MINUTE)
>>>
>>> UNION ALL
>>>
>>> SELECT
>>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>>> sid AS `supply_id`,
>>> 'd84' AS `field_key`,
>>> d84 AS `filed_value`,
>>> count(1) AS `pv`
>>> FROM
>>> baidu_log_view
>>> GROUP BY
>>> sid,
>>> d84,
>>> TUMBLE(event_time, INTERVAL '5' MINUTE)
>>>
>>> UNION ALL
>>>
>>> SELECT
>>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>>> sid AS `supply_id`,
>>> 'd86' AS `field_key`,
>>> d86 AS `field_value`,
>>> count(1) AS `pv`
>>> FROM
>>> baidu_log_view
>>> GROUP BY
>>> sid,
>>> d86,
>>> TUMBLE(event_time, INTERVAL '5' MINUTE)
>>> );
>>>
>>>
>>>
>>> 方案3:
>>>
>>>
>>> INSERT INTO flink_sdk_stats
>>> SELECT
>>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>>> `supply_id`,
>>> `field_key`,
>>> `field_value`,
>>> count(1) AS `pv`
>>> FROM
>>> (
>>> SELECT event_time, sid AS `supply_id`, 'd107' AS `field_key`, d107 AS `field_value` FROM baidu_log_view
>>> UNION ALL
>>> SELECT event_time, sid AS `supply_id`, 'd77' AS `field_key`, d77 AS `field_value` FROM baidu_log_view
>>> UNION ALL
>>> SELECT event_time, sid AS `supply_id`, 'd77' AS `field_key`, d77 AS `field_value` FROM baidu_log_view
>>> UNION ALL
>>> SELECT event_time, sid AS `supply_id`, 'd79' AS `field_key`, d79 AS `field_value` FROM baidu_log_view
>>> UNION ALL
>>> SELECT event_time, sid AS `supply_id`, 'd80' AS `field_key`, d80 AS `field_value` FROM baidu_log_view
>>> UNION ALL
>>> SELECT event_time, sid AS `supply_id`, 'd81' AS `field_key`, d81 AS `field_value` FROM baidu_log_view
>>> UNION ALL
>>> SELECT event_time, sid AS `supply_id`, 'd83' AS `field_key`, d83 AS `field_value` FROM baidu_log_view
>>> UNION ALL
>>> SELECT event_time, sid AS `supply_id`, 'd84' AS `field_key`, d84 AS `field_value` FROM baidu_log_view
>>> UNION ALL
>>> SELECT event_time, sid AS `supply_id`, 'd86' AS `field_key`, d86 AS `field_value` FROM baidu_log_view
>>> )
>>> GROUP BY
>>> `supply_id`, `field_key`, `field_value`, TUMBLE(event_time, INTERVAL '5' MINUTE);
>>>
>>>
>>> 赵一旦 <hi...@gmail.com> 于2020年12月15日周二 下午10:48写道：
>>>
>>>>
>>>> 需要，针对某个表，按照key1（xxx+yyy+ky1），key2（xxx+yyy+ky2），....等多组key统计。其中xxx+yyy为共同字段。目前有如下3种实现我。
>>>> （1）每组key分别统计，分别insert。
>>>> （2）每组key分别统计，然后union结果，然后insert。
>>>> （3）针对表多次select，然后union，然后再基于key统计，然后insert。
>>>> 第三种方案中，会将ky1、ky2这几个不同的字段通过
>>>>
>>>> select 'ky1' as key_name, ky1 as key_value
>>>> union
>>>> select 'ky2' as key_name, ky2 as key_value
>>>>
>>>> 的方式统一为key这个字段，最后通过（xxx+yyy+key_name+key_value）的方式统计。
>>>>
>>>> 目前发现个问题，方案3中，window结点一直没有watermark，导致不发生计算。
>>>>
>>>>
>>>>
>>>>
Re: FlinkSQL多个select group by统计union后输出，或多个select union后在group by统计的区别，以及问题。
Posted by 赵一旦 <hi...@gmail.com>.
有没有人懂啊。今天的新发现如下。
我看了下我的source结点的WEB-UI上展示的那个名字，然后在文本编辑器中划分了下。发现如下。
方案2:
Source: TableSourceScan(table=[[default_catalog, default_database,
baidu_log, watermark=[-(TO_TIMESTAMP(FROM_UNIXTIME(/(CASE(IS NOT
NULL($1), CAST($1):BIGINT NOT NULL, 0:BIGINT), 1000))), 60000:INTERVAL
SECOND)]]], fields=[cid, server_time, d])
-> (
Calc(select=[(d ITEM _UTF-16LE'106') AS $f0, ((d ITEM _UTF-16LE'77')
IS NOT NULL CASE CAST((d ITEM _UTF-16LE'77')) CASE
_UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS $f1,
Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE
CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time]),
Calc(select=[(d ITEM _UTF-16LE'106') AS $f0, ((d ITEM _UTF-16LE'79')
IS NOT NULL CASE CAST((d ITEM _UTF-16LE'79')) CASE
_UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS $f1,
Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE
CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time]),
Calc(select=[(d ITEM _UTF-16LE'106') AS $f0, ((d ITEM _UTF-16LE'80')
IS NOT NULL CASE CAST((d ITEM _UTF-16LE'80')) CASE
_UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS $f1,
Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE
CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time]),
Calc(select=[(d ITEM _UTF-16LE'106') AS $f0, ((d ITEM _UTF-16LE'81')
IS NOT NULL CASE CAST((d ITEM _UTF-16LE'81')) CASE
_UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS $f1,
Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE
CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time]),
Calc(select=[(d ITEM _UTF-16LE'106') AS $f0, ((d ITEM _UTF-16LE'83')
IS NOT NULL CASE CAST((d ITEM _UTF-16LE'83')) CASE
_UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS $f1,
Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE
CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time]),
Calc(select=[(d ITEM _UTF-16LE'106') AS $f0, ((d ITEM _UTF-16LE'84')
IS NOT NULL CASE CAST((d ITEM _UTF-16LE'84')) CASE
_UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS $f1,
Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE
CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time]),
Calc(select=[(d ITEM _UTF-16LE'106') AS $f0, ((d ITEM _UTF-16LE'86')
IS NOT NULL CASE CAST((d ITEM _UTF-16LE'86')) CASE
_UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS $f1,
Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS NOT NULL CASE
CAST(server_time) CASE 0:BIGINT) / 1000)))) AS event_time])
)
方案3:
Source: TableSourceScan(table=[[default_catalog, default_database,
dr1, watermark=[-(TO_TIMESTAMP(FROM_UNIXTIME(/(CASE(IS NOT NULL($1),
CAST($1):BIGINT NOT NULL, 0:BIGINT), 1000))), 60000:INTERVAL
SECOND)]]], fields=[cid, server_time, d])
-> (
Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS
NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS
event_time, (d ITEM _UTF-16LE'106') AS su
pply_id, _UTF-16LE'd107':VARCHAR(4) CHARACTER SET "UTF-16LE" AS
field_key, ((d ITEM _UTF-16LE'107') IS NOT NULL CASE CAST((d ITEM
_UTF-16LE'107')) CASE _UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER
SET "UTF-16LE") AS field_value]),
Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS
NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS
event_time, (d ITEM _UTF-16LE'106') AS supply_id,
_UTF-16LE'd77':VARCHAR(4) CHARACTER SET "UTF-16LE" AS field_key, ((d
ITEM _UTF-16LE'77') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'77')) CASE
_UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS
field_value]),
Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS
NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS
event_time, (d ITEM _UTF-16LE'106') AS supply_id,
_UTF-16LE'd79':VARCHAR(4) CHARACTER SET "UTF-16LE" AS field_key, ((d
ITEM _UTF-16LE'79') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'79')) CASE
_UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS
field_value]),
Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS
NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS
event_time, (d ITEM _UTF-16LE'106') AS supply_id,
_UTF-16LE'd80':VARCHAR(4) CHARACTER SET "UTF-16LE" AS field_key, ((d
ITEM _UTF-16LE'80') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'80')) CASE
_UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS
field_value]),
Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS
NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS
event_time, (d ITEM _UTF-16LE'106') AS supply_id,
_UTF-16LE'd81':VARCHAR(4) CHARACTER SET "UTF-16LE" AS field_key, ((d
ITEM _UTF-16LE'81') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'81')) CASE
_UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS
field_value]),
Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS
NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS
event_time, (d ITEM _UTF-16LE'106') AS supply_id,
_UTF-16LE'd83':VARCHAR(4) CHARACTER SET "UTF-16LE" AS field_key, ((d
ITEM _UTF-16LE'83') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'83')) CASE
_UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS
field_value]),
Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS
NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS
event_time, (d ITEM _UTF-16LE'106') AS supply_id,
_UTF-16LE'd84':VARCHAR(4) CHARACTER SET "UTF-16LE" AS field_key, ((d
ITEM _UTF-16LE'84') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'84')) CASE
_UTF-16LE'NULL':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AS
field_value]),
Calc(select=[Reinterpret(TO_TIMESTAMP(FROM_UNIXTIME(((server_time IS
NOT NULL CASE CAST(server_time) CASE 0:BIGINT) / 1000)))) AS
event_time, (d ITEM _UTF-16LE'106') AS supply_id,
_UTF-16LE'd86':VARCHAR(4) CHARACTER SET "UTF-16LE" AS field_key, ((d
ITEM _UTF-16LE'86') IS NOT NULL CASE CAST((d ITEM _UTF-16LE'86')) CASE
_UTF-16LE'NULL':VARCHAR(214
7483647) CHARACTER SET "UTF-16LE") AS field_value])
)
赵一旦 <hi...@gmail.com> 于2020年12月15日周二 下午10:50写道：
> 方案2没问题，方案3的window算子部分没有watermark。
>
> 赵一旦 <hi...@gmail.com> 于2020年12月15日周二 下午10:49写道：
>
>> 具体SQL如下。
>> 方案2：
>>
>>
>> INSERT INTO flink_sdk_stats
>> (
>> SELECT
>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>> sid AS `supply_id`,
>> 'd77' AS `field_key`,
>> d77 AS `filed_value`,
>> count(1) AS `pv`
>> FROM
>> baidu_log_view
>> GROUP BY
>> sid,
>> d77,
>> TUMBLE(event_time, INTERVAL '5' MINUTE)
>>
>> UNION ALL
>>
>> SELECT
>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>> sid AS `supply_id`,
>> 'd79' AS `field_key`,
>> d79 AS `filed_value`,
>> count(1) AS `pv`
>> FROM
>> baidu_log_view
>> GROUP BY
>> sid,
>> d79,
>> TUMBLE(event_time, INTERVAL '5' MINUTE)
>>
>> UNION ALL
>>
>> SELECT
>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>> sid AS `supply_id`,
>> 'd80' AS `field_key`,
>> d80 AS `filed_value`,
>> count(1) AS `pv`
>> FROM
>> baidu_log_view
>> GROUP BY
>> sid,
>> d80,
>> TUMBLE(event_time, INTERVAL '5' MINUTE)
>>
>> UNION ALL
>>
>> SELECT
>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>> sid AS `supply_id`,
>> 'd81' AS `field_key`,
>> d81 AS `filed_value`,
>> count(1) AS `pv`
>> FROM
>> baidu_log_view
>> GROUP BY
>> sid,
>> d81,
>> TUMBLE(event_time, INTERVAL '5' MINUTE)
>>
>> UNION ALL
>>
>> SELECT
>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>> sid AS `supply_id`,
>> 'd83' AS `field_key`,
>> d83 AS `filed_value`,
>> count(1) AS `pv`
>> FROM
>> baidu_log_view
>> GROUP BY
>> sid,
>> d83,
>> TUMBLE(event_time, INTERVAL '5' MINUTE)
>>
>> UNION ALL
>>
>> SELECT
>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>> sid AS `supply_id`,
>> 'd84' AS `field_key`,
>> d84 AS `filed_value`,
>> count(1) AS `pv`
>> FROM
>> baidu_log_view
>> GROUP BY
>> sid,
>> d84,
>> TUMBLE(event_time, INTERVAL '5' MINUTE)
>>
>> UNION ALL
>>
>> SELECT
>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>> sid AS `supply_id`,
>> 'd86' AS `field_key`,
>> d86 AS `field_value`,
>> count(1) AS `pv`
>> FROM
>> baidu_log_view
>> GROUP BY
>> sid,
>> d86,
>> TUMBLE(event_time, INTERVAL '5' MINUTE)
>> );
>>
>>
>>
>> 方案3:
>>
>>
>> INSERT INTO flink_sdk_stats
>> SELECT
>> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
>> `supply_id`,
>> `field_key`,
>> `field_value`,
>> count(1) AS `pv`
>> FROM
>> (
>> SELECT event_time, sid AS `supply_id`, 'd107' AS `field_key`, d107 AS `field_value` FROM baidu_log_view
>> UNION ALL
>> SELECT event_time, sid AS `supply_id`, 'd77' AS `field_key`, d77 AS `field_value` FROM baidu_log_view
>> UNION ALL
>> SELECT event_time, sid AS `supply_id`, 'd77' AS `field_key`, d77 AS `field_value` FROM baidu_log_view
>> UNION ALL
>> SELECT event_time, sid AS `supply_id`, 'd79' AS `field_key`, d79 AS `field_value` FROM baidu_log_view
>> UNION ALL
>> SELECT event_time, sid AS `supply_id`, 'd80' AS `field_key`, d80 AS `field_value` FROM baidu_log_view
>> UNION ALL
>> SELECT event_time, sid AS `supply_id`, 'd81' AS `field_key`, d81 AS `field_value` FROM baidu_log_view
>> UNION ALL
>> SELECT event_time, sid AS `supply_id`, 'd83' AS `field_key`, d83 AS `field_value` FROM baidu_log_view
>> UNION ALL
>> SELECT event_time, sid AS `supply_id`, 'd84' AS `field_key`, d84 AS `field_value` FROM baidu_log_view
>> UNION ALL
>> SELECT event_time, sid AS `supply_id`, 'd86' AS `field_key`, d86 AS `field_value` FROM baidu_log_view
>> )
>> GROUP BY
>> `supply_id`, `field_key`, `field_value`, TUMBLE(event_time, INTERVAL '5' MINUTE);
>>
>>
>> 赵一旦 <hi...@gmail.com> 于2020年12月15日周二 下午10:48写道：
>>
>>>
>>> 需要，针对某个表，按照key1（xxx+yyy+ky1），key2（xxx+yyy+ky2），....等多组key统计。其中xxx+yyy为共同字段。目前有如下3种实现我。
>>> （1）每组key分别统计，分别insert。
>>> （2）每组key分别统计，然后union结果，然后insert。
>>> （3）针对表多次select，然后union，然后再基于key统计，然后insert。
>>> 第三种方案中，会将ky1、ky2这几个不同的字段通过
>>>
>>> select 'ky1' as key_name, ky1 as key_value
>>> union
>>> select 'ky2' as key_name, ky2 as key_value
>>>
>>> 的方式统一为key这个字段，最后通过（xxx+yyy+key_name+key_value）的方式统计。
>>>
>>> 目前发现个问题，方案3中，window结点一直没有watermark，导致不发生计算。
>>>
>>>
>>>
>>>
Re: FlinkSQL多个select group by统计union后输出，或多个select union后在group by统计的区别，以及问题。
Posted by 赵一旦 <hi...@gmail.com>.
方案2没问题，方案3的window算子部分没有watermark。
赵一旦 <hi...@gmail.com> 于2020年12月15日周二 下午10:49写道：
> 具体SQL如下。
> 方案2：
>
>
> INSERT INTO flink_sdk_stats
> (
> SELECT
> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
> sid AS `supply_id`,
> 'd77' AS `field_key`,
> d77 AS `filed_value`,
> count(1) AS `pv`
> FROM
> baidu_log_view
> GROUP BY
> sid,
> d77,
> TUMBLE(event_time, INTERVAL '5' MINUTE)
>
> UNION ALL
>
> SELECT
> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
> sid AS `supply_id`,
> 'd79' AS `field_key`,
> d79 AS `filed_value`,
> count(1) AS `pv`
> FROM
> baidu_log_view
> GROUP BY
> sid,
> d79,
> TUMBLE(event_time, INTERVAL '5' MINUTE)
>
> UNION ALL
>
> SELECT
> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
> sid AS `supply_id`,
> 'd80' AS `field_key`,
> d80 AS `filed_value`,
> count(1) AS `pv`
> FROM
> baidu_log_view
> GROUP BY
> sid,
> d80,
> TUMBLE(event_time, INTERVAL '5' MINUTE)
>
> UNION ALL
>
> SELECT
> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
> sid AS `supply_id`,
> 'd81' AS `field_key`,
> d81 AS `filed_value`,
> count(1) AS `pv`
> FROM
> baidu_log_view
> GROUP BY
> sid,
> d81,
> TUMBLE(event_time, INTERVAL '5' MINUTE)
>
> UNION ALL
>
> SELECT
> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
> sid AS `supply_id`,
> 'd83' AS `field_key`,
> d83 AS `filed_value`,
> count(1) AS `pv`
> FROM
> baidu_log_view
> GROUP BY
> sid,
> d83,
> TUMBLE(event_time, INTERVAL '5' MINUTE)
>
> UNION ALL
>
> SELECT
> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
> sid AS `supply_id`,
> 'd84' AS `field_key`,
> d84 AS `filed_value`,
> count(1) AS `pv`
> FROM
> baidu_log_view
> GROUP BY
> sid,
> d84,
> TUMBLE(event_time, INTERVAL '5' MINUTE)
>
> UNION ALL
>
> SELECT
> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
> sid AS `supply_id`,
> 'd86' AS `field_key`,
> d86 AS `field_value`,
> count(1) AS `pv`
> FROM
> baidu_log_view
> GROUP BY
> sid,
> d86,
> TUMBLE(event_time, INTERVAL '5' MINUTE)
> );
>
>
>
> 方案3:
>
>
> INSERT INTO flink_sdk_stats
> SELECT
> DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE), 'yyyyMMddHHmm') AS `time`,
> `supply_id`,
> `field_key`,
> `field_value`,
> count(1) AS `pv`
> FROM
> (
> SELECT event_time, sid AS `supply_id`, 'd107' AS `field_key`, d107 AS `field_value` FROM baidu_log_view
> UNION ALL
> SELECT event_time, sid AS `supply_id`, 'd77' AS `field_key`, d77 AS `field_value` FROM baidu_log_view
> UNION ALL
> SELECT event_time, sid AS `supply_id`, 'd77' AS `field_key`, d77 AS `field_value` FROM baidu_log_view
> UNION ALL
> SELECT event_time, sid AS `supply_id`, 'd79' AS `field_key`, d79 AS `field_value` FROM baidu_log_view
> UNION ALL
> SELECT event_time, sid AS `supply_id`, 'd80' AS `field_key`, d80 AS `field_value` FROM baidu_log_view
> UNION ALL
> SELECT event_time, sid AS `supply_id`, 'd81' AS `field_key`, d81 AS `field_value` FROM baidu_log_view
> UNION ALL
> SELECT event_time, sid AS `supply_id`, 'd83' AS `field_key`, d83 AS `field_value` FROM baidu_log_view
> UNION ALL
> SELECT event_time, sid AS `supply_id`, 'd84' AS `field_key`, d84 AS `field_value` FROM baidu_log_view
> UNION ALL
> SELECT event_time, sid AS `supply_id`, 'd86' AS `field_key`, d86 AS `field_value` FROM baidu_log_view
> )
> GROUP BY
> `supply_id`, `field_key`, `field_value`, TUMBLE(event_time, INTERVAL '5' MINUTE);
>
>
> 赵一旦 <hi...@gmail.com> 于2020年12月15日周二 下午10:48写道：
>
>>
>> 需要，针对某个表，按照key1（xxx+yyy+ky1），key2（xxx+yyy+ky2），....等多组key统计。其中xxx+yyy为共同字段。目前有如下3种实现我。
>> （1）每组key分别统计，分别insert。
>> （2）每组key分别统计，然后union结果，然后insert。
>> （3）针对表多次select，然后union，然后再基于key统计，然后insert。
>> 第三种方案中，会将ky1、ky2这几个不同的字段通过
>>
>> select 'ky1' as key_name, ky1 as key_value
>> union
>> select 'ky2' as key_name, ky2 as key_value
>>
>> 的方式统一为key这个字段，最后通过（xxx+yyy+key_name+key_value）的方式统计。
>>
>> 目前发现个问题，方案3中，window结点一直没有watermark，导致不发生计算。
>>
>>
>>
>>
Re: FlinkSQL多个select group by统计union后输出，或多个select union后在group by统计的区别，以及问题。
Posted by 赵一旦 <hi...@gmail.com>.
具体SQL如下。
方案2：
INSERT INTO flink_sdk_stats
(
SELECT
DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE),
'yyyyMMddHHmm') AS `time`,
sid
AS `supply_id`,
'd77'
AS `field_key`,
d77
AS `filed_value`,
count(1)
AS `pv`
FROM
baidu_log_view
GROUP BY
sid,
d77,
TUMBLE(event_time, INTERVAL '5' MINUTE)
UNION ALL
SELECT
DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE),
'yyyyMMddHHmm') AS `time`,
sid
AS `supply_id`,
'd79'
AS `field_key`,
d79
AS `filed_value`,
count(1)
AS `pv`
FROM
baidu_log_view
GROUP BY
sid,
d79,
TUMBLE(event_time, INTERVAL '5' MINUTE)
UNION ALL
SELECT
DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE),
'yyyyMMddHHmm') AS `time`,
sid
AS `supply_id`,
'd80'
AS `field_key`,
d80
AS `filed_value`,
count(1)
AS `pv`
FROM
baidu_log_view
GROUP BY
sid,
d80,
TUMBLE(event_time, INTERVAL '5' MINUTE)
UNION ALL
SELECT
DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE),
'yyyyMMddHHmm') AS `time`,
sid
AS `supply_id`,
'd81'
AS `field_key`,
d81
AS `filed_value`,
count(1)
AS `pv`
FROM
baidu_log_view
GROUP BY
sid,
d81,
TUMBLE(event_time, INTERVAL '5' MINUTE)
UNION ALL
SELECT
DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE),
'yyyyMMddHHmm') AS `time`,
sid
AS `supply_id`,
'd83'
AS `field_key`,
d83
AS `filed_value`,
count(1)
AS `pv`
FROM
baidu_log_view
GROUP BY
sid,
d83,
TUMBLE(event_time, INTERVAL '5' MINUTE)
UNION ALL
SELECT
DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE),
'yyyyMMddHHmm') AS `time`,
sid
AS `supply_id`,
'd84'
AS `field_key`,
d84
AS `filed_value`,
count(1)
AS `pv`
FROM
baidu_log_view
GROUP BY
sid,
d84,
TUMBLE(event_time, INTERVAL '5' MINUTE)
UNION ALL
SELECT
DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE),
'yyyyMMddHHmm') AS `time`,
sid
AS `supply_id`,
'd86'
AS `field_key`,
d86
AS `field_value`,
count(1)
AS `pv`
FROM
baidu_log_view
GROUP BY
sid,
d86,
TUMBLE(event_time, INTERVAL '5' MINUTE)
);
方案3:
INSERT INTO flink_sdk_stats
SELECT
DATE_FORMAT(TUMBLE_END(`event_time`, INTERVAL '5' MINUTE),
'yyyyMMddHHmm') AS `time`,
`supply_id`,
`field_key`,
`field_value`,
count(1) AS `pv`
FROM
(
SELECT event_time, sid AS `supply_id`, 'd107' AS `field_key`,
d107 AS `field_value` FROM baidu_log_view
UNION ALL
SELECT event_time, sid AS `supply_id`, 'd77' AS `field_key`, d77
AS `field_value` FROM baidu_log_view
UNION ALL
SELECT event_time, sid AS `supply_id`, 'd77' AS `field_key`, d77
AS `field_value` FROM baidu_log_view
UNION ALL
SELECT event_time, sid AS `supply_id`, 'd79' AS `field_key`, d79
AS `field_value` FROM baidu_log_view
UNION ALL
SELECT event_time, sid AS `supply_id`, 'd80' AS `field_key`, d80
AS `field_value` FROM baidu_log_view
UNION ALL
SELECT event_time, sid AS `supply_id`, 'd81' AS `field_key`, d81
AS `field_value` FROM baidu_log_view
UNION ALL
SELECT event_time, sid AS `supply_id`, 'd83' AS `field_key`, d83
AS `field_value` FROM baidu_log_view
UNION ALL
SELECT event_time, sid AS `supply_id`, 'd84' AS `field_key`, d84
AS `field_value` FROM baidu_log_view
UNION ALL
SELECT event_time, sid AS `supply_id`, 'd86' AS `field_key`, d86
AS `field_value` FROM baidu_log_view
)
GROUP BY
`supply_id`, `field_key`, `field_value`, TUMBLE(event_time,
INTERVAL '5' MINUTE);
赵一旦 <hi...@gmail.com> 于2020年12月15日周二 下午10:48写道：
>
> 需要，针对某个表，按照key1（xxx+yyy+ky1），key2（xxx+yyy+ky2），....等多组key统计。其中xxx+yyy为共同字段。目前有如下3种实现我。
> （1）每组key分别统计，分别insert。
> （2）每组key分别统计，然后union结果，然后insert。
> （3）针对表多次select，然后union，然后再基于key统计，然后insert。
> 第三种方案中，会将ky1、ky2这几个不同的字段通过
>
> select 'ky1' as key_name, ky1 as key_value
> union
> select 'ky2' as key_name, ky2 as key_value
>
> 的方式统一为key这个字段，最后通过（xxx+yyy+key_name+key_value）的方式统计。
>
> 目前发现个问题，方案3中，window结点一直没有watermark，导致不发生计算。
>
>
>
>