从这2个方案的source结点来看没有太大区别。但问题在于，我从web-ui的metric标签查看outputwatermark的时候。发现方案2中0号并行实例存在8个带有outputwatermark的指标（1个source开头，7个calc开头）。方案3中则只有2个。
赵一旦 <hinobl...@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]) > ) > > > 赵一旦 <hinobl...@gmail.com> 于2020年12月15日周二 下午10:50写道： > >> 方案2没问题，方案3的window算子部分没有watermark。 >> >> 赵一旦 <hinobl...@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); >>> >>> >>> 赵一旦 <hinobl...@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，导致不发生计算。 >>>> >>>> >>>> >>>>
